AI/rocm-systems
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[rocprofiler-compute] Refactor to add type annotation and misc (#787)

Browse Source
This commit is contained in:
xuchen-amd
2025-09-12 13:53:24 -04:00
committed by GitHub
parent 37f8da676a
commit 7ed6000e32
62 changed files with 5145 additions and 4849 deletions
Download Patch File Download Diff File Expand all files Collapse all files
projects/rocprofiler-compute/src/utils/analysis_orm.py
+26 -15
View File
@@ -21,6 +21,7 @@
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
##############################################################################el
from typing import Any, Optional
from sqlalchemy import (
JSON,
@@ -30,21 +31,25 @@ from sqlalchemy import (
Integer,
String,
Text,
TextClause,
create_engine,
func,
select,
text,
)
from sqlalchemy.orm import declarative_base, relationship, sessionmaker
from sqlalchemy.engine import Engine
from sqlalchemy.orm import Session, declarative_base, relationship, sessionmaker
from sqlalchemy.sql import Select
from utils.logger import console_debug, console_error
Base = declarative_base()
PREFIX = "compute_"
SCHEMA_VERSION = "1.0.0"
Base = declarative_base()
class Workload(Base):
__tablename__ = f"{PREFIX}workload"
@@ -162,33 +167,38 @@ class Metadata(Base):
class Database:
_session = None
_session: Optional[Session] = None
_engine: Optional[Engine] = None
@classmethod
def init(cls, db_name):
engine = create_engine(f"sqlite:///{db_name}")
Base.metadata.create_all(engine)
cls._session = sessionmaker(bind=engine)()
def init(cls, db_name: str) -> str:
cls._engine = create_engine(f"sqlite:///{db_name}")
Base.metadata.create_all(cls._engine)
cls._session = sessionmaker(bind=cls._engine)()
console_debug(f"SQLite database initialized with name: {db_name}")
return db_name
@classmethod
def get_session(cls):
def get_session(cls) -> Optional[Session]:
return cls._session
@classmethod
def write(self):
def write(cls) -> None:
if cls._session is None:
console_error("No active database session")
try:
self._session.commit()
cls._session.commit()
except Exception as e:
self._session.rollback()
cls._session.rollback()
console_error(f"Error writing analysis database: {e}")
finally:
self._session.close()
cls._session.close()
cls._session = None
def get_views():
views = {
def get_views() -> list[TextClause]:
views: dict[str, Select[Any]] = {
"kernel_view": select(
Dispatch.kernel_name,
func.count(Dispatch.dispatch_id).label("dispatch_count"),
@@ -207,6 +217,7 @@ def get_views():
Value.value,
).join(Value, Metric.metric_uuid == Value.metric_uuid),
}
return [
text(
f"CREATE VIEW {PREFIX}{view_name} AS "
projects/rocprofiler-compute/src/utils/file_io.py
+184 -175
View File
@@ -27,6 +27,7 @@ import os
import re
from collections import OrderedDict
from pathlib import Path
from typing import Any, Optional
import pandas as pd
import yaml
@@ -39,90 +40,71 @@ from utils.logger import console_debug, console_error, console_log, demarcate
# TODO: use pandas chunksize or dask to read really large csv file
# from dask import dataframe as dd
# the build-in config to list kernel names purpose only
top_stats_build_in_config = {
0: {
"id": 0,
"title": "Top Kernels",
"data source": [{"raw_csv_table": {"id": 1, "source": "pmc_kernel_top.csv"}}],
},
1: {
"id": 1,
"title": "Dispatch List",
"data source": [
{"raw_csv_table": {"id": 2, "source": "pmc_dispatch_info.csv"}}
],
},
}
def load_sys_info(f):
def load_sys_info(f: str) -> pd.DataFrame:
"""
Load sys running info from csv file to a df.
"""
return pd.read_csv(f)
def load_panel_configs(dirs):
def load_panel_configs(
dirs: list[str],
) -> OrderedDict[int, dict[str, Any]]:
"""
Load all panel configs from yaml file.
"""
d = {}
for dir in dirs:
for root, _, files in os.walk(dir):
for f in files:
if f.endswith(".yaml"):
with open(Path(root) / f) as file:
configs: dict[int, dict[str, Any]] = {}
for dir_path in dirs:
for root, _, files in os.walk(dir_path):
for file_name in files:
if file_name.endswith(".yaml"):
with open(Path(root) / file_name) as file:
config_yml = yaml.safe_load(file)
# metric key can be None due to some metric-
# tables not having any metrics
# metric key should be empty dict instead of None
for data_source in config_yml["Panel Config"]["data source"]:
panel_config = config_yml["Panel Config"]
for data_source in panel_config["data source"]:
metric_table = data_source.get("metric_table")
if metric_table and metric_table["metric"] is None:
metric_table["metric"] = {}
d[config_yml["Panel Config"]["id"]] = config_yml["Panel Config"]
configs[panel_config["id"]] = panel_config
# TODO: sort metrics as the header order in case they-
# are not defined in the same order
od = OrderedDict(sorted(d.items()))
# for key, value in od.items():
# print(key, value)
return od
return OrderedDict(sorted(configs.items()))
def load_profiling_config(config_dir):
def load_profiling_config(config_dir: str) -> dict[str, Any]:
"""
Load profiling config from yaml file.
"""
config_path = Path(config_dir) / "profiling_config.yaml"
try:
with open(Path(config_dir).joinpath("profiling_config.yaml")) as file:
prof_config = yaml.safe_load(file)
return prof_config
with open(config_path) as file:
return yaml.safe_load(file) or {}
except FileNotFoundError:
console_log(f"Could not find profiling_config.yaml in {config_dir}")
return dict()
return {}
@demarcate
def create_df_kernel_top_stats(
df_in,
raw_data_dir,
filter_gpu_ids,
filter_dispatch_ids,
filter_nodes,
time_unit,
max_stat_num,
kernel_verbose,
sortby="sum",
):
df_in: dict[str, pd.DataFrame],
raw_data_dir: str,
filter_gpu_ids: Optional[list[str]],
filter_dispatch_ids: Optional[list[str]],
filter_nodes: Optional[str],
time_unit: str,
kernel_verbose: int,
sortby: str = "sum",
) -> None:
"""
Create top stats info by grouping kernels with user's filters.
"""
# NB: think about df = pd.DataFrame(df_in["pmc_perf"].copy())
df = df_in["pmc_perf"]
df = df_in["pmc_perf"].copy()
# Demangle original KernelNames
kernel_name_shortener(df, kernel_verbose)
@@ -139,87 +121,105 @@ def create_df_kernel_top_stats(
if filter_dispatch_ids:
# NB: support ignoring the 1st n dispatched execution by '> n'
# The better way may be parsing python slice string
if ">" in filter_dispatch_ids[0]:
m = re.match(r"\> (\d+)", filter_dispatch_ids[0])
df = df[df["Dispatch_ID"] > int(m.group(1))]
first_filter = filter_dispatch_ids[0]
if first_filter.startswith(">"):
match = re.match(r">\s*(\d+)", first_filter)
if match:
threshold = int(match.group(1))
df = df[df["Dispatch_ID"] > threshold]
else:
df = df.loc[df["Dispatch_ID"].astype(str).isin(filter_dispatch_ids)]
# First, create a dispatches file used to populate global vars
dispatch_info = (
df.loc[:, ["Node", "Dispatch_ID", "Kernel_Name", "GPU_ID"]]
dispatch_columns = (
["Node", "Dispatch_ID", "Kernel_Name", "GPU_ID"]
if "Node" in df.columns
else df.loc[:, ["Dispatch_ID", "Kernel_Name", "GPU_ID"]]
)
dispatch_info.to_csv(
str(Path(raw_data_dir).joinpath("pmc_dispatch_info.csv")), index=False
else ["Dispatch_ID", "Kernel_Name", "GPU_ID"]
)
dispatch_info = df[dispatch_columns]
dispatch_output_path = Path(raw_data_dir) / "pmc_dispatch_info.csv"
dispatch_info.to_csv(dispatch_output_path, index=False)
time_stats = pd.concat(
[df["Kernel_Name"], (df["End_Timestamp"] - df["Start_Timestamp"])],
keys=["Kernel_Name", "ExeTime"],
axis=1,
)
grouped = time_stats.groupby(by=["Kernel_Name"]).agg({
"ExeTime": ["count", "sum", "mean", "median"]
# Calculate execution times
execution_times = df["End_Timestamp"] - df["Start_Timestamp"]
time_stats = pd.DataFrame({
"Kernel_Name": df["Kernel_Name"],
"ExeTime": execution_times,
})
time_unit_str = "(" + time_unit + ")"
grouped.columns = [
x.capitalize() + time_unit_str if x != "count" else x.capitalize()
for x in grouped.columns.get_level_values(1)
]
grouped = time_stats.groupby("Kernel_Name")["ExeTime"].agg([
"count",
"sum",
"mean",
"median",
])
key = "Sum" + time_unit_str
grouped[key] = grouped[key].div(config.TIME_UNITS[time_unit])
key = "Mean" + time_unit_str
grouped[key] = grouped[key].div(config.TIME_UNITS[time_unit])
key = "Median" + time_unit_str
grouped[key] = grouped[key].div(config.TIME_UNITS[time_unit])
# Rename columns with time unit
time_unit_suffix = f"({time_unit})"
column_mapping = {
"count": "Count",
"sum": f"Sum{time_unit_suffix}",
"mean": f"Mean{time_unit_suffix}",
"median": f"Median{time_unit_suffix}",
}
grouped = grouped.rename(columns=column_mapping)
grouped = grouped.reset_index() # Remove special group indexing
# Convert time units
time_divisor = config.TIME_UNITS[time_unit]
for col in [
f"Sum{time_unit_suffix}",
f"Mean{time_unit_suffix}",
f"Median{time_unit_suffix}",
]:
grouped[col] = grouped[col] / time_divisor
key = "Sum" + time_unit_str
grouped["Pct"] = grouped[key] / grouped[key].sum() * 100
grouped = grouped.reset_index()
# Calculate percentage
sum_column = f"Sum{time_unit_suffix}"
grouped["Pct"] = grouped[sum_column] / grouped[sum_column].sum() * 100
# NB:
# Sort by total time as default.
if sortby == "sum":
grouped = grouped.sort_values(by=("Sum" + time_unit_str), ascending=False)
grouped.to_csv(
str(Path(raw_data_dir).joinpath("pmc_kernel_top.csv")), index=False
)
grouped = grouped.sort_values(sum_column, ascending=False)
grouped.to_csv(str(Path(raw_data_dir) / "pmc_kernel_top.csv"), index=False)
elif sortby == "kernel":
grouped = grouped.sort_values("Kernel_Name")
grouped.to_csv(
str(Path(raw_data_dir).joinpath("pmc_kernel_top.csv")), index=False
)
grouped.to_csv(str(Path(raw_data_dir) / "pmc_kernel_top.csv"), index=False)
@demarcate
def create_df_pmc(
raw_data_root_dir, nodes, spatial_multiplexing, kernel_verbose, verbose, config
):
raw_data_root_dir: str,
nodes: Optional[list[str]],
spatial_multiplexing: bool,
kernel_verbose: int,
verbose: int,
config_dict: dict[str, Any],
) -> pd.DataFrame:
"""
Load all raw pmc counters and join into one df.
"""
def create_single_df_pmc(raw_data_dir, node_name, kernel_verbose, verbose):
dfs = []
coll_levels = []
def create_single_df_pmc(
raw_data_dir: str, node_name: Optional[str], kernel_verbose: int, verbose: int
) -> pd.DataFrame:
dfs: list[pd.DataFrame] = []
coll_levels: list[str] = []
df = pd.DataFrame() # noqa: F841
new_df = pd.DataFrame() # noqa: F841
for root, dirs, files in os.walk(raw_data_dir):
for f in files:
# print("file ", f)
if (f.endswith(".csv") and f.startswith("SQ")) or (
f == schema.pmc_perf_file_prefix + ".csv"
):
tmp_df = pd.read_csv(str(Path(root).joinpath(f)))
if config.get("format_rocprof_output") == "rocpd":
for root, _, files in os.walk(raw_data_dir):
for file_name in files:
# Process SQ*.csv or pmc_perf.csv files
is_sq_file = file_name.endswith(".csv") and file_name.startswith("SQ")
is_pmc_perf = file_name == f"{schema.PMC_PERF_FILE_PREFIX}.csv"
if is_sq_file or is_pmc_perf:
file_path = Path(root) / file_name
tmp_df = pd.read_csv(file_path)
if config_dict.get("format_rocprof_output") == "rocpd":
tmp_df = rocpd_data.process_rocpd_csv(tmp_df)
# Demangle original KernelNames
kernel_name_shortener(tmp_df, kernel_verbose)
@@ -228,117 +228,126 @@ def create_df_pmc(
# multiindexing level. Here, we add it into pmc_perf
# as it is the main sub-df which can be handled easily
# later.
if f == "pmc_perf.csv" and node_name != None:
if file_name == "pmc_perf.csv" and node_name is not None:
tmp_df.insert(0, "Node", node_name)
dfs.append(tmp_df)
coll_levels.append(f[:-4])
# Remove .csv extension for collection level
coll_levels.append(file_name[:-4])
if not dfs:
return pd.DataFrame()
# TODO: double check the case if all tmp_df.shape[0] are not on the same page
final_df = pd.concat(dfs, keys=coll_levels, axis=1, join="inner", copy=False)
if verbose >= 2:
console_debug("pmc_raw_data final_single_df %s" % final_df.info)
console_debug(f"pmc_raw_data final_single_df {final_df.info}")
return final_df
root_path = Path(raw_data_root_dir)
# 1. spatial multiplexing case
if spatial_multiplexing:
df = pd.DataFrame()
# todo: more err check
for subdir in Path(raw_data_root_dir).iterdir():
dfs: list[pd.DataFrame] = []
for subdir in root_path.iterdir():
if subdir.is_dir():
new_df = create_single_df_pmc(
subdir, str(subdir.name), kernel_verbose, verbose
str(subdir), str(subdir.name), kernel_verbose, verbose
)
df = pd.concat([df, new_df])
return df
if not new_df.empty:
dfs.append(new_df)
return pd.concat(dfs, ignore_index=True) if dfs else pd.DataFrame()
# specified node list
else:
# regular single node case
if nodes is None:
return create_single_df_pmc(
raw_data_root_dir, None, kernel_verbose, verbose
)
# 2. regular single node case (nodes=None)
if nodes is None:
return create_single_df_pmc(raw_data_root_dir, None, kernel_verbose, verbose)
# "empty list" means all nodes
elif not nodes:
df = pd.DataFrame()
# todo: more err check
for subdir in Path(raw_data_root_dir).iterdir():
if subdir.is_dir():
new_df = create_single_df_pmc(
subdir, str(subdir.name), kernel_verbose, verbose
)
df = pd.concat([df, new_df])
return df
# 3. all nodes case (nodes=[])
if not nodes:
dfs: list[pd.DataFrame] = []
# specified node list
else:
df = pd.DataFrame()
# todo: more err check
for subdir in nodes:
p = Path(raw_data_root_dir)
for subdir in root_path.iterdir():
if subdir.is_dir():
new_df = create_single_df_pmc(
p.joinpath(subdir), subdir, kernel_verbose, verbose
str(subdir), str(subdir.name), kernel_verbose, verbose
)
df = pd.concat([df, new_df])
return df
if not new_df.empty:
dfs.append(new_df)
return pd.concat(dfs, ignore_index=True) if dfs else pd.DataFrame()
# 4. specified node list case (nodes=[...])
dfs: list[pd.DataFrame] = []
for node in nodes:
node_path = root_path / node
if node_path.exists():
new_df = create_single_df_pmc(str(node_path), node, kernel_verbose, verbose)
if not new_df.empty:
dfs.append(new_df)
return pd.concat(dfs, ignore_index=True) if dfs else pd.DataFrame()
def collect_wave_occu_per_cu(in_dir, out_dir, numSE):
def collect_wave_occu_per_cu(in_dir: str, out_dir: str, num_se: int) -> None:
"""
Collect wave occupancy info from in_dir csv files
and consolidate into out_dir/wave_occu_per_cu.csv.
It depends highly on wave_occu_se*.csv format.
"""
in_path = Path(in_dir)
all_data = pd.DataFrame()
all = pd.DataFrame()
for i in range(num_se):
file_path = in_path / f"wave_occu_se{i}.csv"
if not file_path.exists():
continue
for i in range(numSE):
p = Path(in_dir, "wave_occu_se" + str(i) + ".csv")
if p.exists():
tmp_df = pd.read_csv(p)
SE_idx = "SE" + str(tmp_df.loc[0, "SE"])
tmp_df.rename(
columns={
"Dispatch": "Dispatch",
"SE": "SE",
"CU": "CU",
"Occupancy": SE_idx,
},
inplace=True,
)
tmp_df = pd.read_csv(file_path)
if tmp_df.empty:
continue
# TODO: join instead of concat!
if i == 0:
all = tmp_df[{"CU", SE_idx}]
all.sort_index(axis=1, inplace=True)
else:
all = pd.concat([all, tmp_df[SE_idx]], axis=1, copy=False)
se_idx = f"SE{tmp_df.loc[0, 'SE']}"
tmp_df.rename(
columns={
"Dispatch": "Dispatch",
"SE": "SE",
"CU": "CU",
"Occupancy": se_idx,
}
)
if not all.empty:
# print(all.transpose())
all.to_csv(Path(out_dir, "wave_occu_per_cu.csv"), index=False)
# TODO: join instead of concat!
if i == 0:
all_data = tmp_df[{"CU", se_idx}]
all_data.sort_index(axis=1, inplace=True)
else:
all_data = pd.concat([all_data, tmp_df[se_idx]], axis=1, copy=False)
if not all_data.empty:
all_data.to_csv(Path(out_dir) / "wave_occu_per_cu.csv", index=False)
def is_single_panel_config(root_dir, supported_archs):
def is_single_panel_config(
root_dir: str, supported_archs: dict[str, str]
) -> Optional[bool]:
"""
Check the root configs dir structure to decide using one config set for all
archs, or one for each arch.
"""
# If not single config, verify all supported archs have defined configs
supported_archs = supported_archs.keys()
counter = 0
for arch in supported_archs:
if root_dir.joinpath(arch).exists():
counter += 1
if counter == 0:
arch_names = list(supported_archs.keys())
root_path = Path(root_dir)
arch_count = sum(1 for arch in arch_names if (root_path / arch).exists())
if arch_count == 0:
return True
elif counter == len(supported_archs):
elif arch_count == len(arch_names):
return False
else:
console_error("Found multiple panel config sets but incomplete for all archs.")
def find_1st_sub_dir(directory):
def find_1st_sub_dir(directory: str) -> Optional[str]:
"""
Find the first sub dir in a directory
"""
@@ -347,7 +356,7 @@ def find_1st_sub_dir(directory):
# Iterate over entries in the directory
for entry in dir_path.iterdir():
if entry.is_dir(): # Check if it's a directory
return entry
return str(entry)
return None
except FileNotFoundError:
print(f"The directory '{directory}' does not exist.")
return None
console_error(f'The directory "{directory}" does not exist.', exit=False)
projects/rocprofiler-compute/src/utils/gui.py
+235 -220
View File
@@ -23,10 +23,12 @@
##############################################################################
import colorlover
from typing import Any
import colorlover # type: ignore
import pandas as pd
import plotly.express as px
from dash import dash_table, html
import plotly.express as px # type: ignore
from dash import dash_table, html # type: ignore
from utils import schema
from utils.logger import console_error
@@ -41,73 +43,79 @@ IS_DARK = True # TODO: Remove hardcoded in favor of class property
##################
# HELPER FUNCTIONS
##################
def filter_df(column, df, filt):
filt_df = df
if filt != []:
filt_df = df.loc[df[schema.pmc_perf_file_prefix][column].astype(str).isin(filt)]
return filt_df
def filter_df(column: str, df: pd.DataFrame, filt: list[str]) -> pd.DataFrame:
if not filt:
return df
return df.loc[df[schema.PMC_PERF_FILE_PREFIX][column].astype(str).isin(filt)]
def multi_bar_chart(table_id, display_df):
def multi_bar_chart(
table_id: int, display_df: pd.DataFrame
) -> dict[str, dict[str, Any]]:
nested_bar: dict[str, dict[str, Any]] = {}
if table_id == 1604:
nested_bar = {}
for index, row in display_df.iterrows():
if not row["Coherency"] in nested_bar:
nested_bar[row["Coherency"]] = {}
nested_bar[row["Coherency"]][row["Xfer"]] = row["Avg"]
if table_id == 1705: # L2 - Fabric Interface Stalls
nested_bar = {}
for index, row in display_df.iterrows():
if not row["Transaction"] in nested_bar:
nested_bar[row["Transaction"]] = {}
nested_bar[row["Transaction"]][row["Type"]] = row["Avg"]
for _, row in display_df.iterrows():
coherency = row["Coherency"]
if coherency not in nested_bar:
nested_bar[coherency] = {}
nested_bar[coherency][row["Xfer"]] = row["Avg"]
elif table_id == 1705: # L2 - Fabric Interface Stalls
for _, row in display_df.iterrows():
transaction = row["Transaction"]
if transaction not in nested_bar:
nested_bar[transaction] = {}
nested_bar[transaction][row["Type"]] = row["Avg"]
return nested_bar
def discrete_background_color_bins(df, n_bins=5, columns="all"):
def discrete_background_color_bins(
df: pd.DataFrame, n_bins: int = 5, columns: str | list[str] = "all"
) -> tuple[list[dict[str, Any]], html.Div]:
bounds = [i * (1.0 / n_bins) for i in range(n_bins + 1)]
if columns == "all":
if "id" in df:
df_numeric_columns = df.select_dtypes("number").drop(["id"], axis=1)
else:
df_numeric_columns = df.select_dtypes("number")
df_numeric_columns = (
df.select_dtypes("number").drop(["id"], axis=1)
if "id" in df.columns
else df.select_dtypes("number")
)
else:
df_numeric_columns = df[columns]
df_max = df_numeric_columns.max().max()
df_min = df_numeric_columns.min().min()
ranges = [((df_max - df_min) * i) + df_min for i in bounds]
styles = []
legend = []
styles: list[dict[str, Any]] = []
legend: list[html.Div] = []
for i in range(1, len(bounds)):
min_bound = ranges[i - 1]
max_bound = ranges[i]
backgroundColor = colorlover.scales[str(n_bins)]["seq"]["Blues"][i - 1]
background_color = colorlover.scales[str(n_bins)]["seq"]["Blues"][i - 1]
color = "white" if i > len(bounds) / 2.0 else "inherit"
for column in df_numeric_columns:
for column in df_numeric_columns.columns:
filter_query = f"{{{column}}} >= {min_bound}" + (
f" && {{{column}}} < {max_bound}" if i < len(bounds) - 1 else ""
)
styles.append({
"if": {
"filter_query": (
"{{{column}}} >= {min_bound}"
+ (
" && {{{column}}} < {max_bound}"
if (i < len(bounds) - 1)
else ""
)
).format(column=column, min_bound=min_bound, max_bound=max_bound),
"filter_query": filter_query,
"column_id": column,
},
"backgroundColor": backgroundColor,
"backgroundColor": background_color,
"color": color,
})
legend.append(
html.Div(
style={"display": "inline-block", "width": "60px"},
children=[
html.Div(
style={
"backgroundColor": backgroundColor,
"backgroundColor": background_color,
"borderLeft": "1px rgb(50, 50, 50) solid",
"height": "10px",
}
@@ -117,112 +125,84 @@ def discrete_background_color_bins(df, n_bins=5, columns="all"):
)
)
return (styles, html.Div(legend, style={"padding": "5px 0 5px 0"}))
return styles, html.Div(legend, style={"padding": "5px 0 5px 0"})
####################
# GRAPHICAL ELEMENTS
####################
def build_bar_chart(display_df, table_config, barchart_elements, norm_filt):
"""
Read data into a bar chart. ID will determine which subtype of barchart.
"""
d_figs = []
def create_instruction_mix_bar_chart(display_df: pd.DataFrame, df_unit: str) -> px.bar:
display_df = display_df.copy()
display_df["Avg"] = display_df["Avg"].apply(lambda x: int(x) if x != "" else 0)
# Insr Mix bar chart
if table_config["id"] in barchart_elements["instr_mix"]:
display_df["Avg"] = [
x.astype(int) if x != "" else int(0) for x in display_df["Avg"]
]
df_unit = display_df["Unit"].iloc[0]
d_figs.append(
return px.bar(
display_df,
x="Avg",
y="Metric",
color="Avg",
labels={"Avg": f"# of {df_unit.lower()}"},
height=400,
orientation="h",
)
def create_multi_bar_charts(
display_df: pd.DataFrame, table_id: int, df_unit: str
) -> list[px.bar]:
display_df = display_df.copy()
display_df["Avg"] = display_df["Avg"].apply(lambda x: int(x) if x != "" else 0)
nested_bar = multi_bar_chart(table_id, display_df)
charts = []
for group, metric in nested_bar.items():
chart = px.bar(
title=group,
x=list(metric.values()),
y=list(metric.keys()),
labels={"x": df_unit, "y": ""},
text=list(metric.values()),
orientation="h",
height=200,
)
chart.update_xaxes(showgrid=False, rangemode="nonnegative")
chart.update_yaxes(showgrid=False)
chart.update_layout(title_x=0.5)
charts.append(chart)
return charts
def create_sol_charts(display_df: pd.DataFrame, table_id: int) -> list[px.bar]:
display_df = display_df.copy()
display_df["Avg"] = display_df["Avg"].apply(lambda x: float(x) if x != "" else 0.0)
charts = []
if table_id == 1701:
# Special layout for L2 Cache SOL
pct_data = display_df[display_df["Unit"] == "Pct"]
charts.append(
px.bar(
display_df,
pct_data,
x="Avg",
y="Metric",
color="Avg",
labels={"Avg": "# of {}".format(df_unit.lower())},
height=400,
range_color=[0, 100],
labels={"Avg": "%"},
height=220,
orientation="h",
)
).update_xaxes(range=[0, 110], ticks="inside", title="%")
)
# Multi bar chart
elif table_config["id"] in barchart_elements["multi_bar"]:
display_df["Avg"] = [
x.astype(int) if x != "" else int(0) for x in display_df["Avg"]
]
df_unit = display_df["Unit"].iloc[0]
nested_bar = multi_bar_chart(table_config["id"], display_df)
# generate chart for each coherency
for group, metric in nested_bar.items():
d_figs.append(
# HBM Bandwidth chart
hbm_row = display_df[display_df["Metric"] == "HBM Bandwidth"]
if not hbm_row.empty:
hbm_bw = float(hbm_row["Avg"].iloc[0])
gb_data = display_df[display_df["Unit"] == "Gb/s"]
charts.append(
px.bar(
title=group,
x=metric.values(),
y=metric.keys(),
labels={"x": df_unit, "y": ""},
text=metric.values(),
orientation="h",
height=200,
)
.update_xaxes(showgrid=False, rangemode="nonnegative")
.update_yaxes(showgrid=False)
.update_layout(title_x=0.5)
)
# L2 Cache per channel
# elif table_config["id"] in barchart_elements["l2_cache_per_chan"]:
# nested_bar = {}
# channels = []
# for colName, colData in display_df.items():
# if colName == "Channel":
# channels = list(colData.values)
# else:
# display_df[colName] = [
# x.astype(float) if x != "" and x != None else float(0)
# for x in display_df[colName]
# ]
# nested_bar[colName] = list(display_df[colName])
# for group, metric in nested_bar.items():
# d_figs.append(
# px.bar(
# title=group[0 : group.rfind("(")],
# x=channels,
# y=metric,
# labels={
# "x": "Channel",
# "y": group[group.rfind("(") + 1 : len(group) - 1].replace(
# "per", norm_filt
# ),
# },
# ).update_yaxes(rangemode="nonnegative")
# )
# Speed-of-light bar chart
elif table_config["id"] in barchart_elements["sol"]:
display_df["Avg"] = [
float(x) if x != "" else float(0) for x in display_df["Avg"]
]
if table_config["id"] == 1701:
# special layout for L2 Cache SOL
d_figs.append(
px.bar(
display_df[display_df["Unit"] == "Pct"],
x="Avg",
y="Metric",
color="Avg",
range_color=[0, 100],
labels={"Avg": "%"},
height=220,
orientation="h",
).update_xaxes(range=[0, 110], ticks="inside", title="%")
) # append first % chart
hbm_bw = float(
display_df[display_df["Metric"] == "HBM Bandwidth"]["Avg"].iloc[0]
)
d_figs.append(
px.bar(
display_df[display_df["Unit"] == "Gb/s"],
gb_data,
x="Avg",
y="Metric",
color="Avg",
@@ -231,88 +211,145 @@ def build_bar_chart(display_df, table_config, barchart_elements, norm_filt):
height=220,
orientation="h",
).update_xaxes(range=[0, hbm_bw])
) # append second GB/s chart
elif table_config["id"] == 1101:
# Special formatting reference 'Pct of Peak' value
display_df["Pct of Peak"] = [
x.astype(float) if x != "" else float(0)
for x in display_df["Pct of Peak"]
]
d_figs.append(
px.bar(
display_df,
x="Pct of Peak",
y="Metric",
color="Pct of Peak",
range_color=[0, 100],
labels={"Avg": "%"},
height=400,
orientation="h",
).update_xaxes(range=[0, 110])
)
else:
d_figs.append(
px.bar(
display_df,
x="Avg",
y="Metric",
color="Avg",
range_color=[0, 100],
labels={"Avg": "%"},
height=400,
orientation="h",
).update_xaxes(range=[0, 110])
)
elif table_id == 1101:
# Special formatting reference 'Pct of Peak' value
display_df["Pct of Peak"] = display_df["Pct of Peak"].apply(
lambda x: float(x) if x != "" else 0.0
)
charts.append(
px.bar(
display_df,
x="Pct of Peak",
y="Metric",
color="Pct of Peak",
range_color=[0, 100],
labels={"Avg": "%"},
height=400,
orientation="h",
).update_xaxes(range=[0, 110])
)
else:
console_error(
"Table id %s. Cannot determine barchart type." % table_config["id"]
charts.append(
px.bar(
display_df,
x="Avg",
y="Metric",
color="Avg",
range_color=[0, 100],
labels={"Avg": "%"},
height=400,
orientation="h",
).update_xaxes(range=[0, 110])
)
# update layout for each of the charts
for fig in d_figs:
return charts
def build_bar_chart(
display_df: pd.DataFrame,
table_config: dict[str, Any],
barchart_elements: dict[str, Any],
) -> list:
"""
Read data into a bar chart. ID will determine which subtype of barchart.
"""
table_id = table_config["id"]
charts: list[px.bar] = []
# Get unit from first row if available
df_unit = display_df["Unit"].iloc[0] if "Unit" in display_df.columns else ""
# Instruction Mix bar chart
if table_id in barchart_elements["instr_mix"]:
charts.append(create_instruction_mix_bar_chart(display_df, df_unit))
# Multi bar chart
elif table_id in barchart_elements["multi_bar"]:
charts.extend(create_multi_bar_charts(display_df, table_id, df_unit))
# Speed-of-light bar chart
elif table_id in barchart_elements["sol"]:
charts.extend(create_sol_charts(display_df, table_id))
else:
console_error(
f"Table id {table_id}. Cannot determine barchart type.", exit=False
)
return []
# Apply consistent styling to all charts
for fig in charts:
fig.update_layout(
margin=dict(l=50, r=50, b=50, t=50, pad=4),
paper_bgcolor="rgba(0,0,0,0)",
plot_bgcolor="rgba(0,0,0,0)",
font={"color": "#ffffff"},
)
return d_figs
return charts
def get_dark_mode_styles() -> tuple[
dict[str, Any], dict[str, Any], list[dict[str, Any]]
]:
if not IS_DARK:
return {}, {}, []
style_header = {
"backgroundColor": "rgb(30, 30, 30)",
"color": "white",
"fontWeight": "bold",
}
style_data = {
"backgroundColor": "rgb(50, 50, 50)",
"color": "white",
"whiteSpace": "normal",
"height": "auto",
}
style_data_conditional = [
{"if": {"row_index": "odd"}, "backgroundColor": "rgb(60, 60, 60)"}
]
return style_header, style_data, style_data_conditional
def build_table_chart(
display_df, table_config, original_df, display_columns, comparable_columns, decimal
):
display_df: pd.DataFrame,
table_config: dict[str, Any],
original_df: pd.DataFrame,
display_columns: list[str],
comparable_columns: list[str],
decimal: int,
) -> list[dash_table.DataTable]:
"""
Read data into a DashTable
"""
d_figs = []
# build comlumns/header with formatting
formatted_columns = []
for col in display_df.columns:
if (
str(col).lower() == "pct"
or str(col).lower() == "pop"
or str(col).lower() == "percentage"
):
formatted_columns.append(
dict(
id=col,
name=col,
type="numeric",
format={"specifier": ".{}f".format(decimal)},
)
)
col_lower = str(col).lower()
if col_lower in {"pct", "pop", "percentage"}:
formatted_columns.append({
"id": col,
"name": col,
"type": "numeric",
"format": {"specifier": f".{decimal}f"},
})
elif col in comparable_columns:
formatted_columns.append(
dict(
id=col,
name=col,
type="numeric",
format={"specifier": ".{}f".format(decimal)},
)
)
formatted_columns.append({
"id": col,
"name": col,
"type": "numeric",
"format": {"specifier": f".{decimal}f"},
})
else:
formatted_columns.append(dict(id=col, name=col, type="text"))
formatted_columns.append({"id": col, "name": col, "type": "text"})
# tooltip shows only on the 1st col for now if 'Metric Description' available
table_tooltip = (
@@ -326,7 +363,7 @@ def build_table_chart(
),
"type": "markdown",
}
for column, value in row.items()
for column in row.keys()
}
for row in original_df.to_dict("records")
]
@@ -334,6 +371,9 @@ def build_table_chart(
else None
)
# Get styling based on dark mode
style_header, style_data, style_data_conditional = get_dark_mode_styles()
# build data table with columns, tooltip, df and other properties
d_t = dash_table.DataTable(
id=str(table_config["id"]),
@@ -348,36 +388,11 @@ def build_table_chart(
# style cell
style_cell={"maxWidth": "500px"},
# display style
style_header=(
{
"backgroundColor": "rgb(30, 30, 30)",
"color": "white",
"fontWeight": "bold",
}
if IS_DARK
else {}
),
style_data=(
{
"backgroundColor": "rgb(50, 50, 50)",
"color": "white",
"whiteSpace": "normal",
"height": "auto",
}
if IS_DARK
else {}
),
style_data_conditional=(
[
{"if": {"row_index": "odd"}, "backgroundColor": "rgb(60, 60, 60)"},
]
if IS_DARK
else []
),
style_header=style_header,
style_data=style_data,
style_data_conditional=style_data_conditional,
# the df to display
data=display_df.to_dict("records"),
)
# print("DATA: \n", display_df.to_dict('records'))
d_figs.append(d_t)
return d_figs
# print(d_t.columns)
projects/rocprofiler-compute/src/utils/gui_components/header.py
+28 -37
View File
@@ -23,16 +23,19 @@
##############################################################################
from typing import Any, Union
import dash_bootstrap_components as dbc
import pandas as pd
from dash import dcc, html
from utils import schema
avail_normalizations = ["per_wave", "per_cycle", "per_second", "per_kernel"]
AVAIL_NORMALIZATIONS = ["per_wave", "per_cycle", "per_second", "per_kernel"]
# List all the unique column values for desired column in df, 'target_col'
def list_unique(orig_list, is_numeric):
def list_unique(orig_list: list[str], is_numeric: bool) -> list[str]:
list_set = set(orig_list)
unique_list = list(list_set)
if is_numeric:
@@ -40,18 +43,23 @@ def list_unique(orig_list, is_numeric):
return unique_list
def create_span(input):
return {"label": html.Span(str(input), title=str(input)), "value": str(input)}
def create_span(input_value: str) -> dict[str, Union[html.Span, str]]:
return {
"label": html.Span(str(input_value), title=str(input_value)),
"value": str(input_value),
}
def get_header(raw_pmc, input_filters, kernel_names):
kernel_names = list(
map(
str,
raw_pmc[schema.pmc_perf_file_prefix]["Kernel_Name"],
)
)
kernel_names = [x.strip() for x in kernel_names]
def get_header(
raw_pmc: pd.DataFrame, input_filters: dict[str, Any], kernel_names: list[str]
) -> html.Header:
pmc_data = raw_pmc[schema.PMC_PERF_FILE_PREFIX]
kernel_names = [str(name).strip() for name in pmc_data["Kernel_Name"]]
# Extract GPU and Dispatch IDs
gpu_ids = [str(gpu_id) for gpu_id in pmc_data["GPU_ID"]]
dispatch_ids = [str(dispatch_id) for dispatch_id in pmc_data["Dispatch_ID"]]
return html.Header(
id="home",
children=[
@@ -175,7 +183,7 @@ def get_header(raw_pmc, input_filters, kernel_names):
children=["Normalization:"],
),
dcc.Dropdown(
avail_normalizations,
AVAIL_NORMALIZATIONS,
id="norm-filt",
value=input_filters["normalization"],
clearable=False,
@@ -196,14 +204,7 @@ def get_header(raw_pmc, input_filters, kernel_names):
),
dcc.Dropdown(
list_unique(
list(
map(
str,
raw_pmc[
schema.pmc_perf_file_prefix
]["GPU_ID"],
)
),
gpu_ids,
True,
), # list avail gcd ids
id="gcd-filt",
@@ -229,14 +230,7 @@ def get_header(raw_pmc, input_filters, kernel_names):
children=["Dispatch Filter:"],
),
dcc.Dropdown(
list(
map(
str,
raw_pmc[
schema.pmc_perf_file_prefix
]["Dispatch_ID"],
)
),
dispatch_ids,
id="disp-filt",
multi=True,
# default to any dispatch
@@ -282,15 +276,12 @@ def get_header(raw_pmc, input_filters, kernel_names):
children=["Kernels:"],
),
dcc.Dropdown(
list(
map(
create_span,
list_unique(
orig_list=kernel_names,
is_numeric=False,
), # list avail kernel names
[
create_span(name)
for name in list_unique(
kernel_names, False
)
),
],
id="kernel-filt",
multi=True,
value=input_filters["kernel"],
projects/rocprofiler-compute/src/utils/gui_components/memchart.py
+18 -11
View File
@@ -22,29 +22,34 @@
# THE SOFTWARE.
##############################################################################
from typing import Any
from dash import html
from dash_svg import G, Path, Rect, Svg, Text
from utils import schema
from utils.logger import console_error
from utils.utils import format_scientific_notation_if_needed
# Constants for display formatting
DEFAULT_MAX_LENGTH = 6
DEFAULT_PRECISION = 1
DEFAULT_SCIENTIFIC_WIDTH = 8
def insert_chart_data(mem_data, base_data):
def insert_chart_data(mem_data: list[dict[str, Any]], base_data: schema.Workload) -> G:
if len(mem_data) != 1:
console_error("Memory Chart config doesn't follow expected formatting")
table_config = mem_data[0]["metric_table"]
original_df = base_data.dfs[table_config["id"]]
display_columns = original_df.columns.values.tolist().copy()
display_df = original_df[display_columns]
alias = display_df["Metric"].values
values = display_df["Value"].values
memchart_values = {}
memchart_values: dict[str, Any] = {}
for i in range(0, len(alias)):
memchart_values[alias[i]] = values[i]
@@ -521,7 +526,9 @@ def insert_chart_data(mem_data, base_data):
)
def get_memchart(mem_data, base_data):
def get_memchart(
mem_data: list[dict[str, Any]], base_data: schema.Workload
) -> html.Section:
return html.Section(
id="memchart",
children=[
@@ -2039,7 +2046,7 @@ def get_memchart(mem_data, base_data):
)
def format_value_for_display(value, max_length=6):
def format_value_for_display(value: Any, max_length: int = DEFAULT_MAX_LENGTH) -> str: # noqa: ANN401
"""
Format a value (int, float, or str) into a concise string suitable for display.
@@ -2097,8 +2104,8 @@ def format_value_for_display(value, max_length=6):
sci = format_scientific_notation_if_needed(
abs_val,
align=">",
width_align=8,
precision=1,
width_align=DEFAULT_SCIENTIFIC_WIDTH,
precision=DEFAULT_PRECISION,
fmt_type_align="e",
max_length=max_length,
).strip()
@@ -2110,7 +2117,7 @@ def format_value_for_display(value, max_length=6):
value = normal
if is_negative:
value = "-" + value
value = f"-{value}"
else:
value = str(value)
@@ -2123,11 +2130,11 @@ def format_value_for_display(value, max_length=6):
exponent = value[e_index:]
max_mantissa_len = max_length - len(exponent)
if max_mantissa_len < 1:
value = exponent[: max_length - 1] + "…"
value = f"{exponent[: max_length - 1]}…"
else:
truncated_mantissa = mantissa[:max_mantissa_len]
value = truncated_mantissa + exponent
else:
value = value[: max_length - 1] + "…"
value = f"{value[: max_length - 1]}…"
return value
projects/rocprofiler-compute/src/utils/kernel_name_shortener.py
+156 -85
View File
@@ -26,112 +26,183 @@
import re
import subprocess
from pathlib import Path
from typing import Optional
import pandas as pd
from utils.logger import console_debug, console_error, console_log
cache = dict()
# Module-level cache for demangled kernel names
_NAME_CACHE: dict[str, str] = {}
# Constants
# NOTE: c++filt is a Linux-only solution for demangling C++ symbols.
# TODO: We need to think about Windows support in the future.
# Windows equivalent might be undname.exe or using llvm-cxxfilt.
# CONCERN: Using absolute path here is brittle - c++filt location may vary
# across distributions. TODO: Consider using shutil.which() or PATH lookup instead.
CPP_FILT_PATH = "/usr/bin/c++filt"
MAX_SHORTENING_LEVEL = 5
KERNEL_NAME_COLUMNS = ["Kernel_Name", "Name"]
# Note: shortener is now dependent on a rocprof install with llvm
def kernel_name_shortener(df, level):
def shorten_file(df, level):
global cache
def validate_cpp_filt(cpp_filt_path: str = CPP_FILT_PATH) -> bool:
"""Validate that c++filt binary exists and is executable."""
if not Path(cpp_filt_path).is_file():
console_error(
f"Could not resolve c++filt in expected directory: {cpp_filt_path}"
)
return False
return True
column_name = ""
if "Kernel_Name" in df:
column_name = "Kernel_Name"
if "Name" in df:
column_name = "Name"
if column_name == "Kernel_Name" or column_name == "Name":
# loop through all indices
for index in df.index:
original_name = df.loc[index, column_name]
if original_name in cache:
continue
def demangle_kernel_name(original_name: str, cpp_filt_path: str = CPP_FILT_PATH) -> str:
cmd = [cpp_filt_path, original_name]
cmd = [cpp_filt, original_name]
try:
proc = subprocess.Popen(
cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True
)
demangled_name, error = proc.communicate()
proc = subprocess.Popen(
cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE
)
if proc.returncode != 0:
console_error(f"c++filt failed for {original_name}: {error}", exit=False)
return original_name
demangled_name, e = proc.communicate()
demangled_name = str(demangled_name, "UTF-8").strip()
return demangled_name.strip()
# cache miss, add the shortened name to the dictionary
new_name = ""
matches = ""
except (subprocess.SubprocessError, OSError) as e:
console_error(f"Error running c++filt: {e}", exit=False)
return original_name
names_and_args = re.compile(
r"(?P<name>[( )A-Za-z0-9_]+)([ ,*<>()]+)(::)?"
)
# works for name:
# Kokkos::namespace::init_lock_array_kernel_threadid(int) [clone .kd]
if names_and_args.search(demangled_name):
matches = names_and_args.findall(demangled_name)
else:
# Works for first case '__amd_rocclr_fillBuffer.kd'
cache[original_name] = new_name
if new_name == None or new_name == "":
cache[original_name] = demangled_name
continue
def parse_template_depth(text: str, level: int, current_level: int) -> tuple[str, int]:
result = ""
curr_index = 0
current_level = 0
for name in matches:
# can cause errors if a function name-
# or argument is equal to 'clone'
if name[0] == "clone":
continue
if len(name) == 3:
if name[2] == "::":
continue
while curr_index < len(text) and ">" in text:
if current_level < level:
result += text[curr_index:]
current_level -= text[curr_index:].count(">")
break
elif text[curr_index] == ">":
current_level -= 1
curr_index += 1
if current_level < level:
new_name += name[0]
# closing '>' is to be taken account by the while loop
if name[1].count(">") == 0:
if current_level < level:
if not (
current_level == level - 1 and name[1].count("<") > 0
):
new_name += name[1]
current_level += name[1].count("<")
return result, current_level
curr_index = 0
# cases include '>' '> >, ' have to go in depth here to-
# not lose account of commas and current level
while name[1].count(">") > 0 and curr_index < len(name[1]):
if current_level < level:
new_name += name[1][curr_index:]
current_level -= name[1][curr_index:].count(">")
curr_index = len(name[1])
elif name[1][curr_index] == (">"):
current_level -= 1
curr_index += 1
cache[original_name] = new_name
if new_name == None or new_name == "":
cache[original_name] = demangled_name
def shorten_demangled_name(demangled_name: str, level: int) -> str:
names_and_args_pattern = re.compile(r"(?P<name>[( )A-Za-z0-9_]+)([ ,*<>()]+)(::)?")
df[column_name] = df[column_name].map(cache)
matches = names_and_args_pattern.findall(demangled_name)
if not matches:
# Handle cases like '__amd_rocclr_fillBuffer.kd'
return demangled_name
shortened_name = ""
current_level = 0
for name_part, args_part, scope_op in matches:
# Skip 'clone' parts as they can cause errors
if name_part == "clone":
continue
# Skip scope operators
if scope_op == "::":
continue
# Add name part if within level limit
if current_level < level:
shortened_name += name_part
# Handle template arguments
if ">" not in args_part:
if current_level < level:
# Don't add opening brackets at the deepest level
if not (current_level == level - 1 and "<" in args_part):
shortened_name += args_part
current_level += args_part.count("<")
else:
# Handle closing template brackets
if current_level < level:
shortened_name += args_part
current_level -= args_part.count(">")
else:
_, current_level = parse_template_depth(args_part, level, current_level)
return shortened_name if shortened_name else demangled_name
def process_single_kernel_name(
original_name: str, level: int, cpp_filt_path: str = CPP_FILT_PATH
) -> str:
if original_name in _NAME_CACHE:
return _NAME_CACHE[original_name]
demangled_name = demangle_kernel_name(original_name, cpp_filt_path)
shortened_name = shorten_demangled_name(demangled_name, level)
final_name = shortened_name if shortened_name else demangled_name
_NAME_CACHE[original_name] = final_name
return final_name
def get_kernel_column_name(df: pd.DataFrame) -> Optional[str]:
for column_name in KERNEL_NAME_COLUMNS:
if column_name in df.columns:
return column_name
return None
def shorten_file(
df: pd.DataFrame, level: int, cpp_filt_path: str = CPP_FILT_PATH
) -> pd.DataFrame:
column_name = get_kernel_column_name(df)
if not column_name:
console_debug("No kernel name column found")
return df
# Only shorten if valid shortening level
if level < 5:
cpp_filt = str(Path("/usr").joinpath("bin", "c++filt"))
if not Path(cpp_filt).is_file():
console_error(
"Could not resolve c++filt in expected directory: %s" % cpp_filt
)
df_copy = df.copy()
try:
modified_df = shorten_file(df, level)
console_log("profiling", "Kernel_Name shortening complete.")
return modified_df
except pd.errors.EmptyDataError:
console_debug("profiling", "Skipping shortening on empty csv")
df_copy[column_name] = df_copy[column_name].apply(
lambda name: process_single_kernel_name(name, level, cpp_filt_path)
)
return df_copy
def kernel_name_shortener(df: pd.DataFrame, level: int) -> Optional[pd.DataFrame]:
"""Shorten kernel names in a DataFrame.
NOTE: shortener is now dependent on a rocprof install with llvm
Args:
df: DataFrame containing kernel names
level: Shortening level (0-4)
Returns:
DataFrame with shortened kernel names, or None if processing fails
"""
if level >= MAX_SHORTENING_LEVEL:
console_debug("profiling", "Skipping kernel name shortening: level >= 5")
return df
cpp_filt = CPP_FILT_PATH
if not validate_cpp_filt(cpp_filt):
return df
try:
modified_df = shorten_file(df, level, cpp_filt)
console_log("profiling", "Kernel_Name shortening complete.")
return modified_df
except pd.errors.EmptyDataError:
console_debug("profiling", "Skipping shortening on empty csv")
return df
except Exception as e:
console_error(f"Error during kernel name shortening: {e}")
return df
projects/rocprofiler-compute/src/utils/logger.py
+58 -38
View File
@@ -27,6 +27,9 @@ import logging
import os
import sys
from pathlib import Path
from typing import Any, Callable, Optional, TypeVar
R = TypeVar("R")
# Define the colors
BLACK, RED, GREEN, YELLOW, BLUE, MAGENTA, CYAN, WHITE = range(8)
@@ -42,58 +45,80 @@ COLORS = {
"TRACE": MAGENTA,
}
# Constants
TRACE_LEVEL = logging.DEBUG - 5
def demarcate(function):
def wrap_function(*args, **kwargs):
logging.trace("----- [entering function] -> %s()" % (function.__qualname__))
LOG_LEVEL_MAPPING = {
"DEBUG": logging.DEBUG,
"debug": logging.DEBUG,
"TRACE": TRACE_LEVEL,
"trace": TRACE_LEVEL,
"INFO": logging.INFO,
"info": logging.INFO,
"ERROR": logging.ERROR,
"error": logging.ERROR,
}
def demarcate(function: Callable[..., R]) -> Callable[..., R]:
def wrap_function(*args: Any, **kwargs: Any) -> R:
trace_logger(f"----- [entering function] -> {function.__qualname__}()")
result = function(*args, **kwargs)
logging.trace("----- [exiting function] -> %s()" % function.__qualname__)
trace_logger(f"----- [exiting function] -> {function.__qualname__}()")
return result
return wrap_function
def console_error(*argv, exit=True):
def console_error(*argv: Any, exit: bool = True) -> None:
if len(argv) > 1:
logging.error(f"[{argv[0]}] {argv[1]}")
else:
elif len(argv) == 1:
logging.error(f"{argv[0]}")
else:
logging.error("Empty error message")
if exit:
sys.exit(1)
def console_log(*argv, indent_level=0):
def console_log(*argv: Any, indent_level: int = 0) -> None:
indent = ""
if indent_level >= 1:
indent = " " * 3 * indent_level + "|-> " # spaces per indent level
indent = " " * (3 * indent_level) + "|-> " # spaces per indent level
if len(argv) > 1:
logging.info(indent + f"[{argv[0]}] {argv[1]}")
else:
elif len(argv) == 1:
logging.info(indent + f"{argv[0]}")
else:
logging.info(indent + "Empty log message")
def console_debug(*argv):
def console_debug(*argv: Any) -> None:
if len(argv) > 1:
logging.debug(f"[{argv[0]}] {argv[1]}")
else:
elif len(argv) == 1:
logging.debug(f"{argv[0]}")
else:
logging.debug("Empty debug message")
def console_warning(*argv):
def console_warning(*argv: Any) -> None:
if len(argv) > 1:
logging.warning(f"[{argv[0]}] {argv[1]}")
else:
elif len(argv) == 1:
logging.warning(f"{argv[0]}")
else:
logging.warning("Empty warning message")
def trace_logger(message, *args, **kwargs):
logging.log(logging.TRACE, message, *args, **kwargs)
def trace_logger(message: str, *args: Any, **kwargs: Any) -> None:
logging.log(TRACE_LEVEL, message, *args, **kwargs)
# Define the formatter
class ColoredFormatter(logging.Formatter):
def format(self, record):
def format(self, record: logging.LogRecord) -> str:
levelname = record.levelname
if levelname in COLORS:
levelname_color = (
@@ -104,7 +129,7 @@ class ColoredFormatter(logging.Formatter):
class ColoredFormatterAll(logging.Formatter):
def format(self, record):
def format(self, record: logging.LogRecord) -> str:
levelname = record.levelname
if levelname in COLORS:
if levelname == "INFO":
@@ -115,11 +140,12 @@ class ColoredFormatterAll(logging.Formatter):
f"%(levelname)s: %(message)s{RESET_SEQ}"
)
formatter = logging.Formatter(log_fmt)
return formatter.format(record)
return formatter.format(record)
return super().format(record)
class PlainFormatter(logging.Formatter):
def format(self, record):
def format(self, record: logging.LogRecord) -> str:
if record.levelno == logging.ERROR:
self._style._fmt = "%(levelname)s %(message)s"
else:
@@ -129,12 +155,11 @@ class PlainFormatter(logging.Formatter):
# Setup console handler - provided as separate function to be called
# prior to argument parsing
def setup_console_handler():
def setup_console_handler() -> None:
logging.getLogger().handlers.clear()
# register a trace level logger
logging.TRACE = logging.DEBUG - 5
logging.addLevelName(logging.TRACE, "TRACE")
setattr(logging, "TRACE", logging.TRACE)
logging.addLevelName(TRACE_LEVEL, "TRACE")
setattr(logging, "TRACE", TRACE_LEVEL)
setattr(logging, "trace", trace_logger)
color_setting = 1
@@ -164,8 +189,8 @@ def setup_console_handler():
# Setup file handler - enabled in profile mode
def setup_file_handler(loglevel, workload_dir):
filename = str(Path(workload_dir).joinpath("log.txt"))
def setup_file_handler(loglevel: int, workload_dir: str) -> None:
filename = str(Path(workload_dir) / "log.txt")
file_handler = logging.FileHandler(filename, "w")
file_loglevel = min([loglevel, logging.INFO])
file_handler.setLevel(file_loglevel)
@@ -174,9 +199,11 @@ def setup_file_handler(loglevel, workload_dir):
# Setup logger priority - called after argument parsing
def setup_logging_priority(verbosity, quietmode, appmode, guimode):
def setup_logging_priority(
verbosity: int, quietmode: bool, appmode: str, guimode: Optional[bool] = None
) -> int:
# set loglevel based on selected verbosity and quietmode
levels = [logging.INFO, logging.DEBUG, logging.TRACE]
levels = [logging.INFO, logging.DEBUG, TRACE_LEVEL]
if quietmode:
loglevel = logging.ERROR
@@ -191,18 +218,11 @@ def setup_logging_priority(verbosity, quietmode, appmode, guimode):
# optional: override of default loglevel via env variable which takes precedence
if "ROCPROFCOMPUTE_LOGLEVEL" in os.environ.keys():
loglevel = os.environ["ROCPROFCOMPUTE_LOGLEVEL"]
if loglevel in {"DEBUG", "debug"}:
loglevel = logging.DEBUG
elif loglevel in {"TRACE", "trace"}:
loglevel = logging.TRACE
elif loglevel in {"INFO", "info"}:
loglevel = logging.INFO
elif loglevel in {"ERROR", "error"}:
loglevel = logging.ERROR
if loglevel in LOG_LEVEL_MAPPING:
loglevel = LOG_LEVEL_MAPPING[loglevel]
else:
print(
"Ignoring unsupported ROCPROFCOMPUTE_LOGLEVEL setting (%s)" % loglevel
)
print(f"Ignoring unsupported ROCPROFCOMPUTE_LOGLEVEL setting ({loglevel})")
sys.exit(1)
# update console loglevel based on command-line args/env settings
projects/rocprofiler-compute/src/utils/mem_chart.py
+95 -137
View File
@@ -25,14 +25,12 @@
import re
from dataclasses import dataclass, field
from decimal import Decimal
from typing import Dict
from typing import Any, Optional, Union
from plotille import Canvas
from .utils import format_scientific_notation_if_needed
from plotille import Canvas # type: ignore
def make_format_spec(num, align=">"):
def make_format_spec(num: Union[int, float], align: str = ">") -> str:
"""
Generate alignment string for a given input
"""
@@ -45,6 +43,11 @@ def make_format_spec(num, align=">"):
int_part = str(d.to_integral_value())
# Handle special cases where exponent is not an integer (NaN, Infinity, etc.)
if not isinstance(exponent, int):
# For special values, just return basic format
return f"{align}{str(num)}f"
if exponent >= 0:
# Pure integer, or float like 6.0, 6.00 (no decimal places)
if isinstance(num, int):
@@ -60,7 +63,7 @@ def make_format_spec(num, align=">"):
return f"{align}{num_str}f"
def is_value_valid(value):
def is_value_valid(value: Union[int, float, str, None]) -> bool:
"""
Check if a value is valid and display N/A if not
(to be valid, it needs to be not None, and be int or float)
@@ -75,15 +78,15 @@ def is_value_valid(value):
def format_text(
value,
key=None,
value: Union[int, float, str, None],
key: Union[str, Union[int, float], None] = None,
mark_between: str = ": ",
post_description_with_space: str = "",
value_step_prec_rightalign=0,
key_step_prec_leftalign=0,
key_align="<",
value_align=">",
):
value_step_prec_rightalign: Union[int, float] = 0,
key_step_prec_leftalign: Union[int, float] = 0,
key_align: str = "<",
value_align: str = ">",
) -> str:
"""
Format a text string for canvas to display according to
input key-value pair and make proper alignment.
@@ -94,61 +97,25 @@ def format_text(
# Step 1: Build format spec using make_format_spec
value_format = make_format_spec(value_step_prec_rightalign, value_align)
# Step 2: Extract width and precision as integer
match = re.match(r"([<>=^])(\d+)(?:\.(\d+))?([a-zA-Z])?", value_format)
if match:
align_char = match.group(1)
width_align = int(match.group(2))
precision_digits = match.group(3)
fmt_type_align = match.group(4) or "f"
precision = int(precision_digits) if precision_digits else 0
else:
# Fallback to default values
align_char = value_align
width_align = 6
precision = 2
fmt_type_align = "f"
# Step 3: Format the key using make_format_spec
key_format = (
make_format_spec(key_step_prec_leftalign, key_align)
if key is not None
else None
)
key_str = (
"{key:{key_format}}".format(key=key, key_format=key_format)
if key is not None and isinstance(key, (int, float))
else str(key)
if key is not None
else None
)
# Step 4: Format the value or fallback to N/A
if is_value_valid(value):
formatted_value = format_scientific_notation_if_needed(
value,
align=align_char,
width_align=width_align,
precision=precision,
fmt_type_align=fmt_type_align,
max_length=width_align,
sci_lower_bound=1e-3,
sci_upper_bound=1e3,
)
value_str = formatted_value
value_str = f"{value:{value_format}}"
else:
value_str = f"{'N/A':{align_char}{width_align}}"
match = re.search(r"[<>=^](\d+)", value_format)
width = int(match.group(1)) if match else 6
# Step 5: Unit and Final Output
unit_string = post_description_with_space if "N/A" not in value_str else ""
# Use same alignment as in value_format (first char)
align = value_format[0]
value_str = f"{'N/A':{align}{width}}"
if key_str is not None:
if key is not None:
key_format = make_format_spec(key_step_prec_leftalign, key_align)
key_str = f"{key:{key_format}}" if isinstance(key, (int, float)) else str(key)
result_str_no_unit = f"{key_str}{mark_between}{value_str}"
else:
result_str_no_unit = value_str
result_str_no_unit = f"{value_str}"
result_str = result_str_no_unit + unit_string
return result_str
unit_string = post_description_with_space if "N/A" not in value_str else ""
return result_str_no_unit + unit_string
# A basic rect frame for any block or group of wires where all its elements should
@@ -166,11 +133,10 @@ class RectFrame:
# Instr Buff Block
@dataclass
class InstrBuff(RectFrame):
wave_occupancy: int = None
wave_life: int = None
wave_occupancy: Optional[int] = None
wave_life: Optional[int] = None
def draw(self, canvas):
# print("---------", self.x_min, self.y_min, self.x_max, self.y_max)
def draw(self, canvas: Canvas) -> None:
canvas.text(self.x_min, self.y_max + 1.0, self.label)
canvas.rect(self.x_min, self.y_min, self.x_max - 2.0, self.y_max - 1.0)
@@ -208,8 +174,8 @@ class InstrBuff(RectFrame):
# Wires between Instr Buff and Instr Dispatch
@dataclass
class Wire_InstrBuff_InstrDispatch(RectFrame):
def draw(self, canvas):
# Todo: finer wires for connections
def draw(self, canvas: Canvas) -> None:
# TODO: finer wires for connections
canvas.line(self.x_min + 2, self.y_min, self.x_min + 2, self.y_max)
canvas.line(self.x_max, self.y_min + 1.5, self.x_max, self.y_max - 1.5)
canvas.line(self.x_min + 2, self.y_min, self.x_max, self.y_min + 1.5)
@@ -227,9 +193,9 @@ class InstrDispatch(RectFrame):
text_y_offset: float = 0.5
line_y_offset: float = 0.5
rect_y_offset: float = 3.0
instrs: Dict[str, int] = field(default_factory=dict)
instrs: dict[str, int] = field(default_factory=dict)
def draw(self, canvas):
def draw(self, canvas: Canvas) -> None:
canvas.text(self.x_min, self.y_max + 1.0, self.label)
self.top_rect_x_min = self.x_min + 2.0
@@ -237,9 +203,7 @@ class InstrDispatch(RectFrame):
self.top_rect_y_min = self.y_max - 1.5
self.top_rect_y_max = self.y_max
i = 0
for k, v in self.instrs.items():
# print(k,v)
for i, (k, v) in enumerate(self.instrs.items()):
text = format_text(
key=k,
value=v,
@@ -258,7 +222,6 @@ class InstrDispatch(RectFrame):
self.top_rect_y_min - self.rect_y_offset * i,
"------------------>",
)
i = i + 1
# Exec Block
@@ -273,7 +236,7 @@ class Exec(RectFrame):
wavefronts: int = 0
workgroups: int = 0
def draw(self, canvas):
def draw(self, canvas: Canvas) -> None:
canvas.text(self.x_min, self.y_max + 1.0, self.label)
canvas.rect(self.x_min, self.y_min, self.x_max, self.y_max)
@@ -378,13 +341,13 @@ class Exec(RectFrame):
class Wire_E_GLVS(RectFrame):
text_x_offset: float = 3.0
lds_req: int = None
vl1_rd: int = None
vl1_wr: int = None
vl1_atomic: int = None
sl1_rd: int = None
lds_req: Optional[int] = None
vl1_rd: Optional[int] = None
vl1_wr: Optional[int] = None
vl1_atomic: Optional[int] = None
sl1_rd: Optional[int] = None
def draw(self, canvas):
def draw(self, canvas: Canvas) -> None:
canvas.text(
self.x_min + self.text_x_offset,
self.y_max - 2.0,
@@ -459,7 +422,7 @@ class Wire_E_GLVS(RectFrame):
class Wire_InstrBuff_IL1Cache(RectFrame):
il1_fetch: int = 0
def draw(self, canvas):
def draw(self, canvas: Canvas) -> None:
end_col = int(self.y_max - self.y_min)
canvas.text(self.x_min, self.y_max - 1, "^")
for i in range(2, end_col):
@@ -482,10 +445,10 @@ class Wire_InstrBuff_IL1Cache(RectFrame):
# GDS Block
@dataclass
class GDS(RectFrame):
gws: int = None
latency: int = None
gws: Optional[int] = None
latency: Optional[int] = None
def draw(self, canvas):
def draw(self, canvas: Canvas) -> None:
canvas.text(self.x_min, self.y_max + 1.0, self.label)
canvas.rect(self.x_min, self.y_min, self.x_max, self.y_max)
@@ -523,10 +486,10 @@ class GDS(RectFrame):
# LDS Block
@dataclass
class LDS(RectFrame):
util: int = None
latency: int = None
util: Optional[int] = None
latency: Optional[int] = None
def draw(self, canvas):
def draw(self, canvas: Canvas) -> None:
canvas.text(self.x_min, self.y_max + 1.0, self.label)
canvas.rect(self.x_min, self.y_min, self.x_max, self.y_max)
canvas.text(
@@ -556,12 +519,12 @@ class LDS(RectFrame):
# Vector L1 Cache Block
@dataclass
class VectorL1Cache(RectFrame):
hit: int = None
latency: int = None
coales: int = None
stall: int = None
hit: Optional[int] = None
latency: Optional[int] = None
coales: Optional[int] = None
stall: Optional[int] = None
def draw(self, canvas):
def draw(self, canvas: Canvas) -> None:
canvas.text(self.x_min, self.y_max + 1.0, self.label)
canvas.rect(self.x_min, self.y_min, self.x_max, self.y_max)
@@ -614,10 +577,10 @@ class VectorL1Cache(RectFrame):
# Scalar L1D Cache
@dataclass
class ScalarL1DCache(RectFrame):
hit: int = None
latency: int = None
hit: Optional[int] = None
latency: Optional[int] = None
def draw(self, canvas):
def draw(self, canvas: Canvas) -> None:
canvas.text(self.x_min, self.y_max + 1.0, self.label)
canvas.rect(self.x_min, self.y_min, self.x_max, self.y_max)
@@ -648,10 +611,10 @@ class ScalarL1DCache(RectFrame):
# Instr L1 Cache
@dataclass
class InstrL1Cache(RectFrame):
hit: int = None
latency: int = None
hit: Optional[int] = None
latency: Optional[int] = None
def draw(self, canvas):
def draw(self, canvas: Canvas) -> None:
canvas.text(self.x_min, self.y_max + 1.0, self.label)
canvas.rect(self.x_min, self.y_min, self.x_max, self.y_max)
@@ -684,15 +647,15 @@ class InstrL1Cache(RectFrame):
class Wires_L1_L2(RectFrame):
text_v_x_offset: float = 0.0
vl1_l2_rd: int = None
vl1_l2_wr: int = None
vl1_l2_atomic: int = None
sl1_l2_rd: int = None
sl1_l2_wr: int = None
sl1_l2_atomic: int = None
il1_l2_req: int = None
vl1_l2_rd: Optional[int] = None
vl1_l2_wr: Optional[int] = None
vl1_l2_atomic: Optional[int] = None
sl1_l2_rd: Optional[int] = None
sl1_l2_wr: Optional[int] = None
sl1_l2_atomic: Optional[int] = None
il1_l2_req: Optional[int] = None
def draw(self, canvas):
def draw(self, canvas: Canvas) -> None:
canvas.text(
self.x_min + self.text_v_x_offset,
self.y_max - 2.0,
@@ -783,14 +746,14 @@ class Wires_L1_L2(RectFrame):
# L2 Cache
@dataclass
class L2Cache(RectFrame):
rd: int = None
wr: int = None
atomic: int = None
hit: int = None
rd_lat: int = None
wr_lat: int = None
rd: Optional[int] = None
wr: Optional[int] = None
atomic: Optional[int] = None
hit: Optional[int] = None
rd_lat: Optional[int] = None
wr_lat: Optional[int] = None
def draw(self, canvas):
def draw(self, canvas: Canvas) -> None:
canvas.text(self.x_min, self.y_max + 1.0, self.label)
canvas.rect(self.x_min, self.y_min, self.x_max, self.y_max)
@@ -876,11 +839,11 @@ class L2Cache(RectFrame):
class Wire_L2_Fabric(RectFrame):
text_x_offset: float = 3.0
rd: int = None
wr: int = None
atomic: int = None
rd: Optional[int] = None
wr: Optional[int] = None
atomic: Optional[int] = None
def draw(self, canvas):
def draw(self, canvas: Canvas) -> None:
canvas.text(
self.x_min + self.text_x_offset,
self.y_max - 2.0,
@@ -925,7 +888,7 @@ class Wire_L2_Fabric(RectFrame):
# xGMI/PCIe block with wires to fabric
@dataclass
class xGMI_PCIe(RectFrame):
def draw(self, canvas):
def draw(self, canvas: Canvas) -> None:
canvas.rect(self.x_min, self.y_min, self.x_max, self.y_max)
canvas.text(self.x_min + 1.0, self.y_max - 2.0, self.label)
canvas.text(self.x_min + 3.0, self.y_max - 5.0, "^ |")
@@ -937,9 +900,9 @@ class xGMI_PCIe(RectFrame):
# Fabric Cache Block
@dataclass
class Fabric(RectFrame):
lat: Dict[str, int] = field(default_factory=dict)
lat: dict[str, int] = field(default_factory=dict)
def draw(self, canvas):
def draw(self, canvas: Canvas) -> None:
canvas.rect(self.x_min, self.y_min, self.x_max, self.y_max)
canvas.text(self.x_min + 6.0, self.y_max - 2.0, " " + self.label)
canvas.text(self.x_min + 2.0, self.y_max - 4.0, "Latency (cycles)")
@@ -947,24 +910,20 @@ class Fabric(RectFrame):
self.x_min + 2.0, self.y_max - 9, self.x_max - 2.0, self.y_max - 4.5
)
i = 1
for k, v in self.lat.items():
# print(k,v)
for i, (k, v) in enumerate(self.lat.items(), 1):
text = format_text(
key=k,
value=v,
key_step_prec_leftalign=6,
value_step_prec_rightalign=6.0,
)
canvas.text(self.x_min + 4.0, self.y_max - 4.5 - i, text)
i = i + 1
# GMI block with wires to fabric
@dataclass
class GMI(RectFrame):
def draw(self, canvas):
def draw(self, canvas: Canvas) -> None:
canvas.text(self.x_min + 3.0, self.y_max + 4.0, "^ |")
canvas.text(self.x_min + 3.0, self.y_max + 3.0, "| |")
canvas.text(self.x_min + 3.0, self.y_max + 2.0, "| |")
@@ -981,7 +940,7 @@ class Wire_Fabric_HBM(RectFrame):
rd: int = 0
wr: int = 0
def draw(self, canvas):
def draw(self, canvas: Canvas) -> None:
canvas.text(
self.x_min + self.text_x_offset,
self.y_max,
@@ -1013,30 +972,31 @@ class Wire_Fabric_HBM(RectFrame):
# HBM
@dataclass
class HBM(RectFrame):
def draw(self, canvas):
def draw(self, canvas: Canvas) -> None:
canvas.rect(self.x_min, self.y_min, self.x_max, self.y_max)
canvas.text(self.x_min + 4.0, self.y_max - 2.0, self.label)
# Memory chart pannel for 1 instance
class MemChart:
def __init__(self, x_min, y_min, x_max, y_max):
def __init__(self, x_min: float, y_min: float, x_max: float, y_max: float) -> None:
self.x_min = x_min
self.x_max = x_max
self.y_min = y_min
self.y_max = y_max
def draw(self, canvas, normal_unit, metric_dict):
def draw(
self, canvas: Canvas, normal_unit: str, metric_dict: dict[str, Any]
) -> None:
# ----------------------------------------
# Overall rect and title
canvas.rect(self.x_min, self.y_min, self.x_max, self.y_max)
canvas.text(
self.x_min + 2.0, self.y_max - 2.0, "(Normalization: " + normal_unit + ")"
self.x_min + 2.0, self.y_max - 2.0, f"(Normalization: {normal_unit})"
)
# Fixme: this is temp solution to filter out non-numeric string
# FIXME: this is temp solution to filter out non-numeric string
for k, v in metric_dict.items():
# print(k, type(v))
metric_dict[k] = None if isinstance(v, str) else v
# Typically, the drawing order would be: left->right, top->down
@@ -1317,20 +1277,18 @@ class MemChart:
block_hbm.draw(canvas)
def plot_mem_chart(arch, normal_unit, metric_dict):
"""plot memory chart from an arch with given metrics dict"""
def plot_mem_chart(arch: str, normal_unit: str, metric_dict: dict[str, Any]) -> str:
# TODO: verify metrics dict for given arch first
canvas = Canvas(width=234, height=42, xmax=234, ymax=42)
mc = MemChart(0, 0, 233, 41)
mc.draw(canvas, normal_unit, metric_dict)
# return the plot string stream
return canvas.plot()
if __name__ == "__main__":
# TODO: unit test should be moved to tests/*
# Unit test
metric_dict = {}
metric_dict["Wavefront Occupancy"] = 1
projects/rocprofiler-compute/src/utils/mi_gpu_spec.py
+106 -112
View File
@@ -24,60 +24,42 @@
##############################################################################
import os
from dataclasses import dataclass
from typing import Any, Dict
from typing import Any, Optional
import yaml
from utils.logger import console_debug, console_error, console_warning
# Constants for MI series
# NOTE: Currently supports MI50, MI100, MI200, MI300
MI50 = 0
MI100 = 1
MI200 = 2
MI300 = 3
MI350 = 4
MI_CONSTANS = {
MI50: "mi50",
MI100: "mi100",
MI200: "mi200",
MI300: "mi300",
MI350: "mi350",
}
# ----------------------------
# Data Class handling to preserve the hierarchical gpu information
# ----------------------------
@dataclass
class MIGPUSpecs:
_instance = None
_instance: Optional["MIGPUSpecs"] = None
_gpu_series_dict = {} # key: gpu_arch
_gpu_model_dict = {} # key: gpu_arch
_num_xcds_dict = {} # key: gpu_model
_chip_id_dict = {} # key: chip_id (int)
_perfmon_config = {} # key: gpu_arch
_gpu_series_dict: dict[str, str] = {} # key: gpu_arch
_gpu_model_dict: dict[str, list[str]] = {} # key: gpu_arch
_num_xcds_dict: dict[str, dict[str, int]] = {} # key: gpu_model
_chip_id_dict: dict[int, str] = {} # key: chip_id (int)
_perfmon_config: dict[str, Any] = {} # key: gpu_arch
_gpu_arch_to_compute_partition_dict = {} # key: gpu_arch, used for gpu archs
# containing only one gpu model and
# key: gpu_arch, used for gpu archs containing only one gpu model and
# thus one compute partition
_gpu_arch_to_compute_partition_dict: dict[str, dict[str, int]] = {}
_all_gpu_models = []
_all_gpu_models: list[str] = []
_initialized = False
def __new__(cls):
def __new__(cls) -> "MIGPUSpecs":
if cls._instance is None:
cls._instance = super().__new__(cls)
cls._initialize()
return cls._instance
@classmethod
def _initialize(cls):
def _initialize(cls) -> None:
if not cls._initialized:
cls._parse_mi_gpu_spec()
cls._initialized = True
@@ -87,7 +69,7 @@ class MIGPUSpecs:
# ----------------------------
@classmethod
def _load_yaml(cls, file_path: str) -> Dict[str, Any]:
def _load_yaml(cls, file_path: str) -> dict[str, Any]:
"""
Loads MI GPU YAML data /util into a Python dictionary.
@@ -98,23 +80,14 @@ class MIGPUSpecs:
Dict[str, Any]: Parsed YAML data as a nested dictionary.
Exit with console error if an error occurs.
"""
console_debug("[load_yaml]")
try:
with open(file_path, "r") as file:
data = yaml.safe_load(file)
return data
except FileNotFoundError:
console_error(f"Error: The file '{file_path}' was not found.")
except yaml.YAMLError as exc:
console_error(f"Error parsing YAML file '{file_path}': {exc}")
except Exception as e:
console_error(
f"An unexpected error occurred while loading YAML "
f"file '{file_path}': {e}"
)
console_debug("mi_gpu_spec", "[load_yaml]")
with open(file_path) as file:
data = yaml.safe_load(file)
return data or {}
@classmethod
def _parse_mi_gpu_spec(cls):
def _parse_mi_gpu_spec(cls) -> None:
"""
Parse out mi gpu data from yaml file and store in memory.
MI GPUs
@@ -136,16 +109,26 @@ class MIGPUSpecs:
# Load the YAML data
yaml_data = cls._load_yaml(yaml_file_path)
for series in yaml_data["mi_gpu_spec"]:
curr_gpu_series = series["gpu_series"]
console_debug("[parse_mi_gpu_spec] Processing series: %s" % curr_gpu_series)
for archs in series["gpu_archs"]:
curr_gpu_arch = archs["gpu_arch"]
for series in yaml_data.get("mi_gpu_spec", []):
curr_gpu_series = series.get("gpu_series")
if not curr_gpu_series:
continue
console_debug(
"mi_gpu_spec",
f"[parse_mi_gpu_spec] Processing series: {curr_gpu_series}",
)
for archs in series.get("gpu_archs", []):
curr_gpu_arch = archs.get("gpu_arch")
cls._gpu_series_dict[curr_gpu_arch] = curr_gpu_series
cls._perfmon_config[curr_gpu_arch] = archs["perfmon_config"]
cls._perfmon_config[curr_gpu_arch] = archs.get("perfmon_config", {})
cls._gpu_model_dict[curr_gpu_arch] = []
for models in archs["models"]:
for models in archs.get("models", []):
curr_gpu_model = models["gpu_model"]
cls._all_gpu_models.append(curr_gpu_model)
cls._gpu_model_dict[curr_gpu_arch].append(curr_gpu_model)
cls._num_xcds_dict[curr_gpu_model] = (
@@ -153,6 +136,7 @@ class MIGPUSpecs:
.get("compute_partition_mode", {})
.get("num_xcds", {})
)
if "chip_ids" in models and "physical" in models["chip_ids"]:
cls._chip_id_dict[models["chip_ids"]["physical"]] = (
curr_gpu_model
@@ -166,7 +150,7 @@ class MIGPUSpecs:
cls._populate_gpu_arch_to_compute_partition_dict()
@classmethod
def _populate_gpu_arch_to_compute_partition_dict(cls):
def _populate_gpu_arch_to_compute_partition_dict(cls) -> None:
"""
This creates a mapping of gpu_arch -> compute_partition for architectures
where there's only one model (and therefore one partition configuration).
@@ -181,53 +165,55 @@ class MIGPUSpecs:
compute_partition
)
console_debug(
"[populate_single_arch_partition_dict] Single model "
"arch found: %s -> %s (partition: %s)"
% (gpu_arch, single_model, compute_partition)
f"[populate_single_arch_partition_dict] Single model "
f"arch found: {gpu_arch} -> {single_model} (partition:"
f" {compute_partition})"
)
@classmethod
def get_gpu_series_dict(cls):
def get_gpu_series_dict(cls) -> dict[str, str]:
if not cls._gpu_series_dict:
console_error(
"gpu_series_dict not yet populated, did you run parse_mi_gpu_spec()?"
"gpu_series_dict not yet populated, did you run parse_mi_gpu_spec()?",
exit=False,
)
return None
return cls._gpu_series_dict
@classmethod
def get_gpu_series(cls, gpu_arch_):
def get_gpu_series(cls, gpu_arch: str) -> Optional[str]:
if not cls._gpu_series_dict:
console_error(
"gpu_series_dict not yet populated, did you run parse_mi_gpu_spec()?"
"gpu_series_dict not yet populated, did you run parse_mi_gpu_spec()?",
exit=False,
)
return None
# Normalize the key by checking both the raw and lowercase versions
gpu_series = cls._gpu_series_dict.get(gpu_arch_) or cls._gpu_series_dict.get(
gpu_arch_.lower()
gpu_series = cls._gpu_series_dict.get(gpu_arch) or cls._gpu_series_dict.get(
gpu_arch.lower()
)
if gpu_series:
return gpu_series.upper()
console_warning(f"No matching gpu series found for gpu arch: {gpu_arch_}")
console_warning(f"No matching gpu series found for gpu arch: {gpu_arch}")
return None
@classmethod
def get_perfmon_config(cls, gpu_arch_):
def get_perfmon_config(cls, gpu_arch: str) -> dict[Any, Any]:
# Check that gpu_model_dict is populated first
if not cls._perfmon_config:
console_error(
"gpu_model_dict not yet populated. Did you run parse_mi_gpu_spec()?"
)
return None
gpu_arch_lower = gpu_arch_.lower()
return cls._perfmon_config.get(gpu_arch_lower, None)
return cls._perfmon_config.get(gpu_arch.lower(), {})
@classmethod
def get_gpu_model(cls, gpu_arch_, chip_id_):
def get_gpu_model(
cls, gpu_arch: Optional[str], chip_id: Optional[str] = None
) -> Optional[str]:
if not gpu_arch and not chip_id:
return None
# Check that gpu_model_dict is populated first
if not cls._gpu_model_dict:
console_error(
@@ -235,43 +221,55 @@ class MIGPUSpecs:
)
return None
gpu_arch_lower = gpu_arch_.lower()
gpu_arch_lower = gpu_arch.lower()
# Handle gfx942 with chip_id mapping
if gpu_arch_lower not in ("gfx908", "gfx90a"):
if chip_id_ and int(chip_id_) in cls._chip_id_dict:
gpu_model = cls._chip_id_dict.get(int(chip_id_))
if chip_id and chip_id.isdigit():
chip_id_int = int(chip_id)
if chip_id_int in cls._chip_id_dict:
gpu_model = cls._chip_id_dict[chip_id_int]
else:
console_warning(f"No gpu model found for chip id: {chip_id}")
return None
else:
console_warning(f"No gpu model found for chip id: {chip_id_}")
console_warning(f"No valid chip id provided: {chip_id}")
return None
# Otherwise use gpu_model_dict mapping for other mi architectures
elif gpu_arch_lower in cls._gpu_model_dict:
# NOTE: take the first element works for now
gpu_model = cls._gpu_model_dict[gpu_arch_lower][0]
gpu_models = cls._gpu_model_dict[gpu_arch_lower]
if gpu_models:
gpu_model = gpu_models[0]
else:
console_warning(f"No gpu models found for gpu arch: {gpu_arch_lower}")
return None
else:
console_warning(f"No gpu model found for gpu arch: {gpu_arch_lower}")
return None
if not gpu_model:
console_warning(f"No gpu model found for gpu arch: {gpu_arch_lower}")
return None
return gpu_model.upper()
return gpu_model.upper() if gpu_model else None
@classmethod
def set_default_gpu_settings(self, gpu_arch, gpu_model, compute_partition):
def set_default_gpu_settings(
cls,
gpu_arch: Optional[str],
gpu_model: Optional[str],
compute_partition: Optional[str],
) -> int:
"""
Set default GPU settings when model is unknown or cannot be
determined. NOTE: This is a fallback to gfx942 settings -
consider making this architecture-specific.
"""
DEFAULT_COMPUTE_PARTITION = "SPX"
DEFAULT_NUM_XCD = 8
console_warning(
"Unable to determine xcd count from:\n\t"
f"GPU arch: '{gpu_arch}', model: '{gpu_model}',\n\t"
f"partition: '{compute_partition}'"
f'GPU arch: "{gpu_arch}", model: "{gpu_model}",\n\t'
f'partition: "{compute_partition}"'
)
console_warning(
f"Applying default gfx942 settings:\n"
@@ -283,8 +281,11 @@ class MIGPUSpecs:
@classmethod
def get_num_xcds(
cls, gpu_arch: str = None, gpu_model: str = None, compute_partition: str = None
):
cls,
gpu_arch: Optional[str] = None,
gpu_model: Optional[str] = None,
compute_partition: Optional[str] = None,
) -> int:
"""
Retrieve the number of XCDs based on GPU architecture, model,
and compute partition.
@@ -310,22 +311,22 @@ class MIGPUSpecs:
return 1
# 2. Try architecture-based lookup first (preferred method)
if gpu_arch_norm and hasattr(cls, "_gpu_arch_to_compute_partition_dict"):
arch_dict = cls._gpu_arch_to_compute_partition_dict
if gpu_arch_norm in arch_dict:
num_xcds = arch_dict[gpu_arch_norm]
if gpu_arch_norm and gpu_arch_norm in cls._gpu_arch_to_compute_partition_dict:
arch_dict = cls._gpu_arch_to_compute_partition_dict[gpu_arch_norm]
if partition_norm and partition_norm in arch_dict:
num_xcds = arch_dict[partition_norm]
if num_xcds is not None:
return num_xcds
else:
console_warning(
f"No compute partition data found for "
f"architecture '{gpu_arch.upper()}'"
)
else:
console_warning(
f"No compute partition data found for "
f"architecture: {gpu_arch.upper() if gpu_arch else None}"
)
# 3. Fall back to model + partition-based lookup
if gpu_model_norm:
# Validate XCD dictionary is populated
if not hasattr(cls, "_num_xcds_dict") or not cls._num_xcds_dict:
if not cls._num_xcds_dict:
console_error(
"mi300_num_xcds_dict not populated. "
"Did you run parse_mi_gpu_spec()?"
@@ -343,9 +344,8 @@ class MIGPUSpecs:
)
elif partition_norm not in model_dict:
console_warning(
f"Unknown compute partition "
f"'{compute_partition}' for model "
f"'{gpu_model}'"
f"Unknown compute partition: "
f"{compute_partition} for model: {gpu_model}"
)
else:
num_xcds = model_dict[partition_norm]
@@ -364,25 +364,19 @@ class MIGPUSpecs:
return cls.set_default_gpu_settings(gpu_arch, gpu_model, compute_partition)
@classmethod
def get_chip_id_dict(cls):
if cls._chip_id_dict:
return cls._chip_id_dict
else:
console_error()
def get_chip_id_dict(cls) -> dict[int, str]:
return cls._chip_id_dict
@classmethod
def get_num_xcds_dict(cls):
if cls._num_xcds_dict:
return cls._num_xcds_dict
else:
console_error()
def get_num_xcds_dict(cls) -> dict[str, dict[str, int]]:
return cls._num_xcds_dict
@classmethod
def get_gpu_arch_to_compute_partition_dict(cls):
def get_gpu_arch_to_compute_partition_dict(cls) -> dict[str, dict[str, int]]:
return cls._gpu_arch_to_compute_partition_dict
@classmethod
def get_all_gpu_models(cls):
def get_all_gpu_models(cls) -> list:
return cls._all_gpu_models
projects/rocprofiler-compute/src/utils/parser.py
+653 -733
View File
File diff suppressed because it is too large Load Diff
projects/rocprofiler-compute/src/utils/rocpd_data.py
+12 -6
View File
@@ -26,6 +26,9 @@
import csv
import sqlite3
from contextlib import closing
from typing import Any
import pandas as pd
from utils.logger import console_error
@@ -70,19 +73,22 @@ def convert_db_to_csv(
])
for row in cursor:
writer.writerow(row)
except (sqlite3.DatabaseError, IOError) as e:
console_error(f"Error converting database to CSV: {e}")
except OSError as e:
console_error(f"Database error while converting to CSV: {e}")
except Exception as e:
console_error(f"Unexpected error converting database to CSV: {e}")
def process_rocpd_csv(df):
def process_rocpd_csv(df: pd.DataFrame) -> pd.DataFrame:
"""
Merge counters across unique dispatches from the
input dataframe and return processed dataframe.
"""
# Only import pandas if needed
import pandas as pd
if df.empty:
return df
data: list[dict[str, Any]] = []
data = list()
# Group by unique kernel and merge into a single row
for _, group_df in df.groupby([
"Dispatch_ID",
projects/rocprofiler-compute/src/utils/roofline_calc.py
+238 -216
View File
@@ -27,18 +27,18 @@
import csv
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Union
import pandas as pd
from utils import schema
from utils.logger import console_debug, console_warning
from utils.parser import apply_filters, eval_metric
from utils.specs import MachineSpecs
################################################
# Global vars
################################################
IMGNAME = "empirRoof"
XMIN = 0.01
XMAX = 1000
@@ -48,7 +48,7 @@ FONT_WEIGHT = "bold"
# SUPPORTED_DATATYPES table is based on datatype support in rocm-amdgpu-bench repository
# Indicates which datatypes per gpu arch can be generated by the roofline binary
SUPPORTED_DATATYPES = {
SUPPORTED_DATATYPES: dict[str, list[str]] = {
"gfx90a": [
"FP16",
"BF16",
@@ -135,7 +135,49 @@ class AI_Data:
avgDuration: float
def get_font():
@dataclass
class PlotPoints:
"""Data structure for storing roofline plot points."""
ai_l1: list[list[float]]
ai_l2: list[list[float]]
ai_hbm: list[list[float]]
kernelNames: list[str]
@classmethod
def empty(cls) -> "PlotPoints":
"""Create empty plot points structure."""
return cls(ai_l1=[[], []], ai_l2=[[], []], ai_hbm=[[], []], kernelNames=[])
@dataclass
class GraphPoints:
"""Data structure for storing roofline graph ceiling points."""
hbm: list[Union[list[float], float, None]]
l2: list[Union[list[float], float, None]]
l1: list[Union[list[float], float, None]]
lds: list[Union[list[float], float, None]]
valu: list[Union[list[float], float, None]]
mfma: list[Union[list[float], float, None]]
@classmethod
def empty(cls) -> "GraphPoints":
"""Create empty graph points structure."""
return cls(
hbm=[None, None, None],
l2=[None, None, None],
l1=[None, None, None],
lds=[None, None, None],
valu=[None, None, None],
mfma=[None, None, None],
)
################################################
# Helper functions
################################################
def get_font() -> dict[str, Union[int, str]]:
return {
"size": FONT_SIZE,
"color": FONT_COLOR,
@@ -144,160 +186,164 @@ def get_font():
}
def get_color(catagory):
if catagory == "ai_l1":
return "green"
elif catagory == "ai_l2":
return "blue"
elif catagory == "ai_hbm":
return "red"
else:
raise RuntimeError("Invalid catagory passed to get_color()")
def get_color(category: str) -> str:
color_map = {"ai_l1": "green", "ai_l2": "blue", "ai_hbm": "red"}
if category not in color_map:
raise RuntimeError(f"Invalid category passed to get_color(): {category}")
return color_map[category]
# -------------------------------------------------------------------------------------
# Plot BW at each cache level
# -------------------------------------------------------------------------------------
def calc_ceilings(roofline_parameters, dtype, benchmark_data):
def calc_ceilings(
roofline_parameters: dict[str, Any],
dtype: str,
benchmark_data: dict[str, list[str]],
) -> dict[str, list[Union[list[float], float, None]]]:
"""Given benchmarking data, calculate ceilings (or peak performance) for
empirical roofline"""
# TODO: This is where filtering by memory level will need to occur for standalone
graphPoints = {"hbm": [], "l2": [], "l1": [], "lds": [], "valu": [], "mfma": []}
graph_points: dict[str, list[Union[list[float], float, None]]] = {
"hbm": [],
"l2": [],
"l1": [],
"lds": [],
"valu": [],
"mfma": [],
}
if roofline_parameters["mem_level"] == "ALL":
cacheHierarchy = CACHE_HIERARCHY
else:
cacheHierarchy = roofline_parameters["mem_level"]
cache_hierarchy = (
CACHE_HIERARCHY
if roofline_parameters["mem_level"] == "ALL"
else roofline_parameters["mem_level"]
)
x1 = y1 = x2 = y2 = -1
x1_mfma = y1_mfma = x2_mfma = y2_mfma = -1
ops_flops = "Ops" if (dtype[:1] == "I") else "Flops"
ops_flops = "Ops" if dtype.startswith("I") else "Flops"
peak_ops = 0.0
if dtype in PEAK_OPS_DATATYPES:
peakOps = float(
benchmark_data[dtype + "{}".format(ops_flops)][
roofline_parameters["device_id"]
]
peak_ops = float(
benchmark_data[f"{dtype}{ops_flops}"][roofline_parameters["device_id"]]
)
for i in range(0, len(cacheHierarchy)):
for cache_level in cache_hierarchy:
# Plot BW line
console_debug("roofline", "Current cache level is %s" % cacheHierarchy[i])
curr_bw = cacheHierarchy[i] + "Bw"
peakBw = float(benchmark_data[curr_bw][roofline_parameters["device_id"]])
console_debug("roofline", f"Current cache level is {cache_level}")
curr_bw = f"{cache_level}Bw"
peak_bw = float(benchmark_data[curr_bw][roofline_parameters["device_id"]])
x1 = float(XMIN)
y1 = float(XMIN) * peakBw
y1 = float(XMIN) * peak_bw
if dtype in PEAK_OPS_DATATYPES:
x2 = peakOps / peakBw
y2 = peakOps # noqa
x2 = peak_ops / peak_bw
y2 = peak_ops # noqa
# Plot MFMA lines (NOTE: Assuming MI200 soc)
x1_mfma = peakOps / peakBw
y1_mfma = peakOps
x1_mfma = peak_ops / peak_bw
y1_mfma = peak_ops
peak_mfma = 0.0
if dtype in MFMA_DATATYPES:
target_precision = (dtype) if (dtype[:1] == "I") else ("F" + dtype[2:])
target_precision = dtype if dtype.startswith("I") else f"F{dtype[2:]}"
peakMFMA = float(
benchmark_data["MFMA{}{}".format(target_precision, ops_flops)][
peak_mfma = float(
benchmark_data[f"MFMA{target_precision}{ops_flops}"][
roofline_parameters["device_id"]
]
)
x2_mfma = peakMFMA / peakBw
y2_mfma = peakMFMA
x2_mfma = peak_mfma / peak_bw
y2_mfma = peak_mfma
# Check which peak is higher for formatting bandwidth lines
if y2_mfma > y1_mfma: # peakMFMA
peakX = x2_mfma
peakY = y2_mfma
if y2_mfma > y1_mfma: # peak_mfma
peak_x = x2_mfma
peak_y = y2_mfma
else: # peakVALU
peakX = x1_mfma
peakY = y1_mfma
peak_x = x1_mfma
peak_y = y1_mfma
# These are the points to use:
console_debug("roofline", "coordinate points:")
console_debug("x = [{}, {}]".format(x1, peakX))
console_debug("y = [{}, {}]".format(y1, peakY))
console_debug(f"x = [{x1}, {peak_x}]")
console_debug(f"y = [{y1}, {peak_y}]")
graphPoints[cacheHierarchy[i].lower()].append([x1, peakX])
graphPoints[cacheHierarchy[i].lower()].append([y1, peakY])
graphPoints[cacheHierarchy[i].lower()].append(peakBw)
cache_key = cache_level.lower()
graph_points[cache_key].extend([[x1, peak_x], [y1, peak_y], peak_bw])
# ----------------------------------------------------------------------------------
# Plot computing roof
# ----------------------------------------------------------------------------------
if dtype in PEAK_OPS_DATATYPES:
# Plot FMA roof
x0 = XMAX
if x2 < x0:
x0 = x2
x0 = min(x2, XMAX) if x2 < XMAX else XMAX
console_debug("FMA ROOF [{}, {}], [{},{}]".format(x0, XMAX, peakOps, peakOps))
graphPoints["valu"].append([x0, XMAX])
graphPoints["valu"].append([peakOps, peakOps])
graphPoints["valu"].append(peakOps)
console_debug(f"FMA ROOF [{x0}, {XMAX}], [{peak_ops},{peak_ops}]")
graph_points["valu"].extend([[x0, XMAX], [peak_ops, peak_ops], peak_ops])
# Plot MFMA roof
if dtype in MFMA_DATATYPES: # assert that mfma has been assigned
x0_mfma = XMAX
if x2_mfma < x0_mfma:
x0_mfma = x2_mfma
x0_mfma = min(x2_mfma, XMAX) if x2_mfma < XMAX else XMAX
console_debug(
"MFMA ROOF [{}, {}], [{},{}]".format(x0_mfma, XMAX, peakMFMA, peakMFMA)
)
graphPoints["mfma"].append([x0_mfma, XMAX])
graphPoints["mfma"].append([peakMFMA, peakMFMA])
graphPoints["mfma"].append(peakMFMA)
console_debug(f"MFMA ROOF [{x0_mfma}, {XMAX}], [{peak_mfma},{peak_mfma}]")
graph_points["mfma"].extend([
[x0_mfma, XMAX],
[peak_mfma, peak_mfma],
peak_mfma,
])
return graphPoints
return graph_points
# -------------------------------------------------------------------------------------
# Overlay application performance
# -------------------------------------------------------------------------------------
# Calculate relevant metrics for ai calculation
def calc_ai_analyze(workload, mspec, sort_type, config, arch_config):
def calc_ai_analyze(
workload: schema.Workload,
mspec: MachineSpecs,
sort_type: str,
config: dict[str, Any],
arch_config: schema.ArchConfig,
) -> dict[str, Union[list[list[float]], list[str]]]:
"""
Calculate per-kernel metrics and AI points with Roofline yamls using eval_metric.
"""
console_debug("calc_ai_analyze: Starting calc_ai analysis using Roofline yamls")
plot_points = {
"ai_l1": [[], []],
"ai_l2": [[], []],
"ai_hbm": [[], []],
"kernelNames": [],
}
console_debug("calc_ai_analyze", "Starting calc_ai analysis using Roofline yamls")
plot_points = PlotPoints.empty()
workload.roofline_metrics = {}
filtered_pmc = apply_filters(workload, workload.path, is_gui=False, debug=False)
kernel_ids_to_process = []
kernel_ids_to_process: list[int] = []
kernel_top_table_id = 1
if workload.filter_kernel_ids:
kernel_ids_to_process = workload.filter_kernel_ids
else:
if kernel_top_table_id in workload.dfs:
kernel_top_df = workload.dfs[kernel_top_table_id]
kernel_ids_to_process = kernel_top_df.index.tolist()
console_debug(
"roofline", f"Found {len(kernel_ids_to_process)} kernels to process"
)
elif kernel_top_table_id in workload.dfs:
kernel_top_df = workload.dfs[kernel_top_table_id]
kernel_ids_to_process = kernel_top_df.index.tolist()
console_debug(
"roofline", f"Found {len(kernel_ids_to_process)} kernels to process"
)
if not kernel_ids_to_process:
console_warning("No kernels found to process for roofline")
return plot_points
return plot_points.__dict__
for kernel_id in kernel_ids_to_process:
kernel_name = ""
if kernel_top_table_id in workload.dfs:
kernel_top_df = workload.dfs[kernel_top_table_id]
if kernel_id in kernel_top_df.index:
kernel_name = kernel_top_df.loc[kernel_id, "Kernel_Name"]
else:
if kernel_id not in kernel_top_df.index:
continue
kernel_name = kernel_top_df.loc[kernel_id, "Kernel_Name"]
else:
continue
@@ -314,8 +360,8 @@ def calc_ai_analyze(workload, mspec, sort_type, config, arch_config):
kernel_only_data = {"pmc_perf": kernel_pmc_df["pmc_perf"]}
kernel_dfs = {}
kernel_dfs_type = {}
kernel_dfs: dict[int, pd.DataFrame] = {}
kernel_dfs_type: dict[int, str] = {}
for table_id in [401, 402]:
if table_id in arch_config.dfs:
@@ -368,16 +414,16 @@ def calc_ai_analyze(workload, mspec, sort_type, config, arch_config):
# add to plot points if we have valid data
if performance > 0:
if ai_hbm > 0:
plot_points["ai_hbm"][0].append(ai_hbm)
plot_points["ai_hbm"][1].append(performance)
plot_points.ai_hbm[0].append(ai_hbm)
plot_points.ai_hbm[1].append(performance)
if ai_l2 > 0:
plot_points["ai_l2"][0].append(ai_l2)
plot_points["ai_l2"][1].append(performance)
plot_points.ai_l2[0].append(ai_l2)
plot_points.ai_l2[1].append(performance)
if ai_l1 > 0:
plot_points["ai_l1"][0].append(ai_l1)
plot_points["ai_l1"][1].append(performance)
plot_points.ai_l1[0].append(ai_l1)
plot_points.ai_l1[1].append(performance)
plot_points["kernelNames"].append(f"K{kernel_id}")
plot_points.kernelNames.append(f"K{kernel_id}")
console_debug("roofline", f"Added kernel {kernel_id} to plot points")
else:
console_debug(
@@ -391,14 +437,14 @@ def calc_ai_analyze(workload, mspec, sort_type, config, arch_config):
"calc_table": kernel_dfs.get(402, pd.DataFrame()),
}
console_debug(
"roofline", f"Generated {len(plot_points['kernelNames'])} plot points"
)
console_debug("roofline", f"Generated {len(plot_points.kernelNames)} plot points")
console_debug("roofline", f"Plot points: {plot_points}")
return plot_points
return plot_points.__dict__
def calc_ai_profile(mspec, sort_type, ret_df):
def calc_ai_profile(
mspec: MachineSpecs, sort_type: str, ret_df: dict[str, pd.DataFrame]
) -> dict[str, Union[list[list[float]], list[str]]]:
"""Given counter data, calculate arithmetic intensity for each kernel
in the application. Leverage hard-coded equations to calculate AI values.
@@ -410,8 +456,7 @@ def calc_ai_profile(mspec, sort_type, ret_df):
)
df = ret_df["pmc_perf"]
# Sort by top kernels or top dispatches?
df = df.sort_values(by=["Kernel_Name"])
df = df.reset_index(drop=True)
df = df.sort_values(by=["Kernel_Name"]).reset_index(drop=True)
total_flops = valu_flops = mfma_flops_f6f4 = mfma_flops_f8 = mfma_flops_bf16 = (
mfma_flops_f16
@@ -419,25 +464,24 @@ def calc_ai_profile(mspec, sort_type, ret_df):
L2cache_data
) = hbm_data = calls = totalDuration = avgDuration = 0.0
kernelName = ""
kernel_name = ""
my_list: list[AI_Data] = []
myList = []
at_end = False
next_kernelName = ""
supported_dt = SUPPORTED_DATATYPES[mspec.gpu_arch]
supported_dt = (
SUPPORTED_DATATYPES[mspec.gpu_arch]
if mspec.gpu_arch in SUPPORTED_DATATYPES
else None
)
for idx in df.index:
# CASE: Top kernels
# Calculate + append AI data if
# a) current KernelName is different than previous OR
# b) We've reached the end of list
if idx + 1 == df.shape[0]:
at_end = True
else:
next_kernelName = df["Kernel_Name"][idx + 1]
at_end = idx + 1 == df.shape[0]
next_kernel_name = df["Kernel_Name"][idx + 1] if not at_end else ""
kernel_name = df["Kernel_Name"][idx]
kernelName = df["Kernel_Name"][idx]
try:
total_flops += (
(
@@ -476,10 +520,10 @@ def calc_ai_profile(mspec, sort_type, ret_df):
total_flops += df["SQ_INSTS_VALU_MFMA_MOPS_F8"][idx] * 512
if ("FP4" in supported_dt) or ("FP6" in supported_dt):
total_flops += df["SQ_INSTS_VALU_MFMA_MOPS_F6F4"][idx] * 512
except KeyError:
except KeyError as e:
console_debug(
"roofline",
"{}: Skipped total_flops at index {}".format(kernelName[:35], idx),
f"{kernel_name[:35]}: Skipped total_flops at index {idx} due to {e}",
)
pass
try:
@@ -506,10 +550,10 @@ def calc_ai_profile(mspec, sort_type, ret_df):
+ df["SQ_INSTS_VALU_TRANS_F64"][idx]
)
)
except KeyError:
except KeyError as e:
console_debug(
"roofline",
"{}: Skipped valu_flops at index {}".format(kernelName[:35], idx),
f"{kernel_name[:35]}: Skipped valu_flops at index {idx} due to {e}",
)
pass
@@ -523,10 +567,10 @@ def calc_ai_profile(mspec, sort_type, ret_df):
mfma_flops_f32 += df["SQ_INSTS_VALU_MFMA_MOPS_F32"][idx] * 512
mfma_flops_f64 += df["SQ_INSTS_VALU_MFMA_MOPS_F64"][idx] * 512
mfma_iops_i8 += df["SQ_INSTS_VALU_MFMA_MOPS_I8"][idx] * 512
except KeyError:
except KeyError as e:
console_debug(
"roofline",
"{}: Skipped mfma ops at index {}".format(kernelName[:35], idx),
f"{kernel_name[:35]}: Skipped mfma ops at index {idx} due to {e}",
)
pass
@@ -536,19 +580,19 @@ def calc_ai_profile(mspec, sort_type, ret_df):
* 4
* (mspec.lds_banks_per_cu)
)
except KeyError:
except KeyError as e:
console_debug(
"roofline",
"{}: Skipped lds_data at index {}".format(kernelName[:35], idx),
f"{kernel_name[:35]}: Skipped lds_data at index {idx} due to {e}",
)
pass
try:
L1cache_data += df["TCP_TOTAL_CACHE_ACCESSES_sum"][idx] * 64
except KeyError:
except KeyError as e:
console_debug(
"roofline",
"{}: Skipped L1cache_data at index {}".format(kernelName[:35], idx),
f"{kernel_name[:35]}: Skipped L1cache_data at index {idx} due to {e}",
)
pass
@@ -559,10 +603,10 @@ def calc_ai_profile(mspec, sort_type, ret_df):
+ df["TCP_TCC_ATOMIC_WITHOUT_RET_REQ_sum"][idx] * 64
+ df["TCP_TCC_READ_REQ_sum"][idx] * 64
)
except KeyError:
except KeyError as e:
console_debug(
"roofline",
"{}: Skipped L2cache_data at index {}".format(kernelName[:35], idx),
f"{kernel_name[:35]}: Skipped L2cache_data at index {idx} due to {e}",
)
pass
try:
@@ -602,22 +646,21 @@ def calc_ai_profile(mspec, sort_type, ret_df):
)
+ (df["TCC_EA0_WRREQ_64B_sum"][idx] * 64)
)
except KeyError:
except KeyError as e:
console_debug(
"roofline",
"{}: Skipped hbm_data at index {}".format(kernelName[:35], idx),
f"{kernel_name[:35]}: Skipped hbm_data at index {idx} due to {e}",
)
pass
totalDuration += df["End_Timestamp"][idx] - df["Start_Timestamp"][idx]
avgDuration += df["End_Timestamp"][idx] - df["Start_Timestamp"][idx]
calls += 1
if sort_type == "kernels" and (at_end or (kernelName != next_kernelName)):
myList.append(
if sort_type == "kernels" and (at_end or (kernel_name != next_kernel_name)):
my_list.append(
AI_Data(
kernelName,
kernel_name,
calls,
total_flops / calls,
valu_flops / calls,
@@ -636,11 +679,8 @@ def calc_ai_profile(mspec, sort_type, ret_df):
avgDuration / calls,
)
)
console_debug(
"Just added {} to AI_Data at index {}. # of calls: {}".format(
kernelName, idx, calls
)
)
console_debug(f"Just added {kernel_name} to AI_Data. # of calls: {calls}")
total_flops = valu_flops = mfma_flops_f6f4 = mfma_flops_f8 = (
mfma_flops_bf16
) = mfma_flops_f16 = mfma_iops_i8 = mfma_flops_f32 = mfma_flops_f64 = (
@@ -650,9 +690,9 @@ def calc_ai_profile(mspec, sort_type, ret_df):
) = 0.0
if sort_type == "dispatches":
myList.append(
my_list.append(
AI_Data(
kernelName,
kernel_name,
calls,
total_flops,
valu_flops,
@@ -679,73 +719,64 @@ def calc_ai_profile(mspec, sort_type, ret_df):
avgDuration
) = 0.0
myList.sort(key=lambda x: x.totalDuration, reverse=True)
my_list.sort(key=lambda x: x.totalDuration, reverse=True)
intensities = {"ai_l1": [], "ai_l2": [], "ai_hbm": []}
curr_perf = []
kernelNames = []
i = 0
# Create list of top 5 intensities
while i < TOP_N and i != len(myList):
if myList[i].total_flops == 0:
intensities: dict[str, list[float]] = {"ai_l1": [], "ai_l2": [], "ai_hbm": []}
curr_perf: list[float] = []
kernel_names: list[str] = []
# Create list of top N intensities
for i in range(min(TOP_N, len(my_list))):
kernel_data = my_list[i]
if my_list[i].total_flops == 0:
console_debug(
f"No flops counted for {myList[i].KernelName}, "
f"No flops counted for {my_list[i].KernelName}, "
"arithmetic intensities will not display on plots."
)
kernelNames.append(myList[i].KernelName)
(
intensities["ai_l1"].append(myList[i].total_flops / myList[i].L1cache_data)
if myList[i].L1cache_data
else intensities["ai_l1"].append(0)
kernel_names.append(my_list[i].KernelName)
# Calculate arithmetic intensities
intensities["ai_l1"].append(
kernel_data.total_flops / kernel_data.L1cache_data
if kernel_data.L1cache_data
else 0
)
# print("cur_ai_L1", myList[i].total_flops/myList[i].L1cache_data) if myList[i].L1cache_data else print("null") #noqa
# print()
(
intensities["ai_l2"].append(myList[i].total_flops / myList[i].L2cache_data)
if myList[i].L2cache_data
else intensities["ai_l2"].append(0)
intensities["ai_l2"].append(
kernel_data.total_flops / kernel_data.L2cache_data
if kernel_data.L2cache_data
else 0
)
# print("cur_ai_L2", myList[i].total_flops/myList[i].L2cache_data) if myList[i].L2cache_data else print("null") #noqa
# print()
(
intensities["ai_hbm"].append(myList[i].total_flops / myList[i].hbm_data)
if myList[i].hbm_data
else intensities["ai_hbm"].append(0)
intensities["ai_hbm"].append(
kernel_data.total_flops / kernel_data.hbm_data
if kernel_data.hbm_data
else 0
)
# print("cur_ai_hbm", myList[i].total_flops/myList[i].hbm_data) if myList[i].hbm_data else print("null") #noqa
# print()
(
curr_perf.append(myList[i].total_flops / myList[i].avgDuration)
if myList[i].avgDuration
else curr_perf.append(0)
curr_perf.append(
kernel_data.total_flops / kernel_data.avgDuration
if kernel_data.avgDuration
else 0
)
# print("cur_perf", myList[i].total_flops/myList[i].avgDuration) if myList[i].avgDuration else print("null") #noqa
i += 1
# Create intensity points for plotting
intensity_points: dict[str, Union[list[list[float]], list[str]]] = {}
intensityPoints = {"ai_l1": [], "ai_l2": [], "ai_hbm": []}
for ai_type in intensities:
values = intensities[ai_type]
for i in intensities:
values = intensities[i]
x = values
y = curr_perf[: len(values)]
intensity_points[ai_type] = [x, y]
color = get_color(i) # noqa
x = []
y = []
for entryIndx in range(0, len(values)):
x.append(values[entryIndx])
y.append(curr_perf[entryIndx])
intensityPoints[i].append(x)
intensityPoints[i].append(y)
# Add an entry for kernel names
intensityPoints["kernelNames"] = kernelNames
return intensityPoints
# Add kernel names
intensity_points["kernelNames"] = kernel_names
return intensity_points
def constuct_roof(roofline_parameters, dtype):
def construct_roof(
roofline_parameters: dict[str, Any], dtype: str
) -> dict[str, list[Union[list[float], float, None]]]:
workload_dir = roofline_parameters.get("workload_dir")
if isinstance(workload_dir, list):
base_dir = (
@@ -756,44 +787,35 @@ def constuct_roof(roofline_parameters, dtype):
else:
base_dir = workload_dir
benchmark_results = str(Path(base_dir) / "roofline.csv")
benchmark_results = Path(base_dir) / "roofline.csv"
# -----------------------------------------------------
# Initialize roofline data dictionary from roofline.csv
# -----------------------------------------------------
# TODO: consider changing this to an ordered dict for consistency over py versions
benchmark_data = {}
headers = []
benchmark_data: dict[str, list[str]] = {}
headers: list[str] = []
try:
with open(benchmark_results, "r") as csvfile:
csvReader = csv.reader(csvfile, delimiter=",")
rowCount = 0
for row in csvReader:
with open(benchmark_results) as csvfile:
csv_reader = csv.reader(csvfile, delimiter=",")
row_count = 0
for row in csv_reader:
row.pop(0) # remove devID
if rowCount == 0:
if row_count == 0:
headers = row
for i in headers:
benchmark_data[i] = []
for header in headers:
benchmark_data[header] = []
else:
for i, key in enumerate(headers):
benchmark_data[key].append(row[i])
rowCount += 1
csvfile.close()
except Exception:
graphPoints = {
"hbm": [None, None, None],
"l2": [None, None, None],
"l1": [None, None, None],
"lds": [None, None, None],
"valu": [None, None, None],
"mfma": [None, None, None],
}
return graphPoints
row_count += 1
except Exception as e:
console_debug("roofline", f"Failed to read benchmark results: {e}")
return GraphPoints.empty().__dict__
# ------------------
# Generate Roofline
# ------------------
results = calc_ceilings(roofline_parameters, dtype, benchmark_data)
return results
return calc_ceilings(roofline_parameters, dtype, benchmark_data)
projects/rocprofiler-compute/src/utils/schema.py
+21 -17
View File
@@ -30,7 +30,7 @@
from collections import OrderedDict
from dataclasses import dataclass, field
from typing import Dict, List
from typing import Any
import pandas as pd
@@ -38,10 +38,10 @@ import pandas as pd
@dataclass
class ArchConfig:
# [id: panel_config] pairs
panel_configs: OrderedDict = field(default=dict)
panel_configs: OrderedDict[int, Any] = field(default_factory=OrderedDict)
# [id: df] pairs
dfs: Dict[int, pd.DataFrame] = field(default_factory=dict)
dfs: dict[int, pd.DataFrame] = field(default_factory=dict)
# NB:
# dfs_type should be a meta info embeded into df.
@@ -49,30 +49,34 @@ class ArchConfig:
# So do it as below for now.
# [id: df_type] pairs
dfs_type: Dict[int, str] = field(default_factory=dict)
dfs_type: dict[int, str] = field(default_factory=dict)
# [Index: Metric name] pairs
metric_list: Dict[str, str] = field(default_factory=dict)
metric_list: dict[str, str] = field(default_factory=dict)
# [Metric name: Counters] pairs
metric_counters: Dict[str, list] = field(default_factory=dict)
metric_counters: dict[str, list] = field(default_factory=dict)
@dataclass
class Workload:
sys_info: pd.DataFrame = None
raw_pmc: pd.DataFrame = None
dfs: Dict[int, pd.DataFrame] = field(default_factory=dict)
dfs_type: Dict[int, str] = field(default_factory=dict)
filter_kernel_ids: List[int] = field(default_factory=list)
filter_gpu_ids: List[int] = field(default_factory=list)
filter_dispatch_ids: List[int] = field(default_factory=list)
filter_nodes: List[str] = field(default_factory=list)
avail_ips: List[int] = field(default_factory=list)
sys_info: pd.DataFrame = field(default_factory=pd.DataFrame)
raw_pmc: pd.DataFrame = field(default_factory=pd.DataFrame)
dfs: dict[int, pd.DataFrame] = field(default_factory=dict)
dfs_type: dict[int, str] = field(default_factory=dict)
filter_kernel_ids: list[int] = field(default_factory=list)
filter_gpu_ids: list[int] = field(default_factory=list)
filter_dispatch_ids: list[int] = field(default_factory=list)
filter_nodes: list[str] = field(default_factory=list)
avail_ips: list[int] = field(default_factory=list)
roofline_peaks: pd.DataFrame = field(default_factory=pd.DataFrame)
roofline_metrics: dict[int, dict[str, Any]] = field(default_factory=dict)
path: str = field(default_factory=str)
filter_top_n: str = field(default_factory=str)
# Metrics will be calculated ONLY when the header(key) is in below list
supported_field = [
SUPPORTED_FIELD = [
"Value",
"Minimum",
"Maximum",
@@ -121,4 +125,4 @@ supported_field = [
]
# The prefix of raw pmc_perf.csv
pmc_perf_file_prefix = "pmc_perf"
PMC_PERF_FILE_PREFIX = "pmc_perf"
projects/rocprofiler-compute/src/utils/specs.py
+400 -258
View File
File diff suppressed because it is too large Load Diff
projects/rocprofiler-compute/src/utils/tty.py
+416 -408
View File
@@ -23,41 +23,46 @@
##############################################################################
import argparse
import copy
import textwrap
from pathlib import Path
from typing import Any, Optional, TextIO
import pandas as pd
from tabulate import tabulate
import config
from utils import mem_chart, parser
from utils import mem_chart, parser, schema
from utils.kernel_name_shortener import kernel_name_shortener
from utils.logger import console_error, console_log, console_warning
from utils.utils import convert_metric_id_to_panel_info, get_uuid
def string_multiple_lines(source, width, max_rows):
def string_multiple_lines(source: str, width: int, max_rows: int) -> str:
"""
Adjust string with multiple lines by inserting '\n'
"""
idx = 0
lines = []
while idx < len(source) and len(lines) < max_rows:
lines.append(source[idx : idx + width])
idx += width
lines: list[str] = []
for i in range(0, len(source), width):
if len(lines) >= max_rows:
break
lines.append(source[i : i + width])
if len(lines) == max_rows and len(source) > max_rows * width:
lines[-1] = lines[-1][:-3] + "..."
if idx < len(source):
last = lines[-1]
lines[-1] = last[0:-3] + "..."
return "\n".join(lines)
def get_table_string(df, transpose=False, decimal=2):
def get_table_string(
df: pd.DataFrame, transpose: bool = False, decimal: int = 2
) -> str:
"""
Convert DataFrame to a formatted table string, wrapping specified columns.
"""
df_to_show = df.transpose() if transpose else df
wrap_columns = ["Description"]
wrap_width = 40
for col in wrap_columns:
@@ -67,42 +72,40 @@ def get_table_string(df, transpose=False, decimal=2):
.astype(str)
.apply(lambda x: textwrap.fill(x, width=wrap_width))
)
df_with_index = df_to_show.reset_index()
return tabulate(
df_to_show,
headers="keys",
df_with_index.values,
headers=list(df_with_index.columns),
tablefmt="fancy_grid",
floatfmt="." + str(decimal) + "f",
floatfmt=f".{decimal}f",
)
def convert_time_columns(df, time_unit):
def convert_time_columns(df: pd.DataFrame, time_unit: str) -> pd.DataFrame:
"""
Convert time column values based on the specified time unit.
Uses the Unit column to identify which columns contain time data.
"""
if time_unit not in config.TIME_UNITS or "Unit" not in df.columns:
return df
# Avoid modifying the original
df_copy = df.copy()
time_rows = df_copy["Unit"].str.lower().str.contains("ns", na=False)
time_value_columns = ["Avg", "Min", "Max"]
for col in time_value_columns:
if col in df_copy.columns:
mask = time_rows
if mask.any():
try:
numeric_values = pd.to_numeric(
df_copy.loc[mask, col], errors="coerce"
)
df_copy.loc[mask, col] = (
numeric_values / config.TIME_UNITS[time_unit]
)
except Exception:
pass
if col in df_copy.columns and time_rows.any():
try:
numeric_values = pd.to_numeric(
df_copy.loc[time_rows, col], errors="coerce"
)
df_copy.loc[time_rows, col] = (
numeric_values / config.TIME_UNITS[time_unit]
)
except Exception:
pass
# Update the Unit column
if time_rows.any():
@@ -111,31 +114,341 @@ def convert_time_columns(df, time_unit):
return df_copy
def has_time_data(df):
def has_time_data(df: pd.DataFrame) -> bool:
"""
Check if the dataframe contains time data by looking at the Unit column.
"""
if "Unit" not in df.columns:
return False
# NOTE: "ns" / "NS" / "nS" / "Ns" are reserved for Nanosec time unit
return df["Unit"].str.lower().str.contains("ns", na=False).any()
return bool(df["Unit"].str.lower().str.contains("ns", na=False).any())
def show_all(args, runs, archConfigs, output, profiling_config, roof_plot=None):
def is_roofline_shown(
args: argparse.Namespace,
runs: dict[str, Any],
output: Optional[TextIO],
panel: dict[str, Any],
roof_plot: Optional[str],
hidden_cols: list[str],
) -> bool:
has_roofline_style = any(
data_source.get(table_type, {}).get("cli_style") == "Roofline"
for data_source in panel["data source"]
for table_type in data_source
)
if not has_roofline_style or (
args.filter_metrics and "4" not in args.filter_metrics
):
return False
print(f"\n{'=' * 80}", file=output)
print("4. Roofline", file=output)
print("=" * 80, file=output)
# Display roofline metrics for each run
for run_path, workload in runs.items():
if hasattr(workload, "roofline_metrics") and workload.roofline_metrics:
print(
"\n(4.1) Per-Kernel Roofline Metrics and (4.2) AI Plot Points",
file=output,
)
print("-" * 80, file=output)
kernel_top_df = workload.dfs.get(1, pd.DataFrame())
if not kernel_top_df.empty:
kernel_name_shortener(kernel_top_df, args.kernel_verbose)
# Display roofline metrics
for kernel_id, metrics in workload.roofline_metrics.items():
if not kernel_top_df.empty and kernel_id in kernel_top_df.index:
kernel_name = kernel_top_df.loc[kernel_id, "Kernel_Name"]
kernel_pct = (
kernel_top_df.loc[kernel_id, "Pct"]
if "Pct" in kernel_top_df.columns
else 0
)
else:
kernel_name = metrics.get("name", f"Kernel {kernel_id}")
kernel_pct = 0
display_name = (
kernel_name[:80] + "..." if len(kernel_name) > 80 else kernel_name
)
print(
f"\nKernel {kernel_id}: {display_name} ({kernel_pct:.1f}%)",
file=output,
)
base_indent = " "
table_indent_prefix = f"{base_indent}| "
print(f"{base_indent}|", file=output)
tables = {
401: (
"4.1 Roofline Rate Metrics:",
metrics.get("ai_table", pd.DataFrame()),
),
402: (
"4.2 Roofline AI Plot Points:",
metrics.get("calc_table", pd.DataFrame()),
),
}
for table_id, (table_name, df) in tables.items():
if df.empty:
continue
print(f"{base_indent}├─ {table_name}", file=output)
# Remove hidden columns
display_df = df.copy()
for col in hidden_cols:
if col in display_df.columns:
display_df = display_df.drop(columns=[col])
table_string = get_table_string(
display_df, transpose=False, decimal=args.decimal
)
indented_table = textwrap.indent(table_string, table_indent_prefix)
print(indented_table, file=output)
else:
print("\nNo per-kernel metrics available", file=output)
# Show the roofline plot
if roof_plot:
show_roof_plot(roof_plot)
return True
def process_table_data(
args: argparse.Namespace,
runs: dict[str, Any],
table_config: dict[str, Any],
table_type: str,
comparable_columns: list[str],
hidden_cols: list[str],
) -> pd.DataFrame:
# take the 1st run as baseline
base_run, base_data = next(iter(runs.items()))
base_df = base_data.dfs[table_config["id"]]
if args.time_unit and has_time_data(base_df):
base_df = convert_time_columns(base_df, args.time_unit)
result_df = pd.DataFrame(index=base_df.index)
for header in base_df.columns:
# Skip filtered columns
if (
table_type != "raw_csv_table"
and args.cols
and base_df.columns.get_loc(header) not in args.cols
):
continue
if header in hidden_cols:
continue
if header not in comparable_columns:
# Process columns that are not comparable across runs.
if (
table_type == "raw_csv_table"
and table_config["source"]
in ["pmc_kernel_top.csv", "pmc_dispatch_info.csv"]
and header == "Kernel_Name"
):
# NB: the width of kernel name might depend
# on the header of the table.
width = 40 if table_config["source"] == "pmc_kernel_top.csv" else 80
max_rows = 3 if table_config["source"] == "pmc_kernel_top.csv" else 4
adjusted_names = base_df["Kernel_Name"].apply(
lambda x: string_multiple_lines(x, width, max_rows)
)
result_df = pd.concat([result_df, adjusted_names], axis=1)
elif table_type == "raw_csv_table" and header == "Info":
for run_data in runs.values():
cur_df = run_data.dfs[table_config["id"]]
result_df = pd.concat([result_df, cur_df[header]], axis=1)
else:
result_df = pd.concat([result_df, base_df[header]], axis=1)
else:
# Process columns that can be compared across runs.
for run_name, run_data in runs.items():
cur_df = run_data.dfs[table_config["id"]]
if args.time_unit and has_time_data(base_df):
cur_df = convert_time_columns(cur_df, args.time_unit)
if (table_type == "raw_csv_table") or (
table_type == "metric_table" and header not in hidden_cols
):
if run_name != base_run:
# Calculate percentage difference between current and
# base dataframe.
base_series = pd.to_numeric(
base_df[header], errors="coerce"
).fillna(0.0)
cur_series = pd.to_numeric(
cur_df[header], errors="coerce"
).fillna(0.0)
# Calculate absolute and percentage differences
absolute_diff = (cur_series - base_series).round(args.decimal)
percentage_diff = (
absolute_diff / base_series.replace(0, 1) * 100
).round(args.decimal)
if args.verbose >= 2:
console_log("---------", header, percentage_diff)
# Format as "value (percentage%)"
formatted_diff = (
cur_series.round(args.decimal).astype(str)
+ " ("
+ percentage_diff.astype(str)
+ "%)"
)
result_df = pd.concat([result_df, formatted_diff], axis=1)
# DEBUG: When in a CI setting and flag is set,
# then verify metrics meet threshold
# requirement
if (
header in ["Value", "Count", "Avg"]
and percentage_diff.abs().gt(args.report_diff).any()
):
result_df["Abs Diff"] = absolute_diff
if args.report_diff:
violation_idx = percentage_diff.index[
percentage_diff.abs() > args.report_diff
]
console_warning(
f"Dataframe diff exceeds {args.report_diff}% "
"threshold requirement\n"
f"See metric {violation_idx.to_numpy()}"
)
console_warning(result_df)
else:
# Base run - just add the rounded values
cur_df_copy = copy.deepcopy(cur_df)
cur_df_copy[header] = [
(round(float(x), args.decimal) if x != "" else x)
for x in base_df[header]
]
result_df = pd.concat([result_df, cur_df_copy[header]], axis=1)
return result_df
def format_table_output(
args: argparse.Namespace,
table_config: dict[str, Any],
df: pd.DataFrame,
table_type: str,
runs: dict[str, Any],
csv_dir: Optional[Path] = None,
) -> str:
"""Format table for output, handling special cases and saving to files if needed."""
table_id_str = f"{table_config['id'] // 100}.{table_config['id'] % 100}"
content = ""
# Check if any column in df is empty
is_empty_columns_exist = any(
df.replace("", None).iloc[:, col_idx].isnull().all()
for col_idx in range(len(df.columns))
)
# Do not print the table if any column is empty
if is_empty_columns_exist:
title = table_config.get("title", "")
console_log(f"Not showing table with empty column(s): {table_id_str} {title}")
return content
if "title" in table_config and table_config["title"]:
content += f"{table_id_str} {table_config['title']}\n"
if args.output_format == "csv" and csv_dir and csv_dir.is_dir():
if "title" in table_config and table_config["title"]:
table_id_str += f"_{table_config['title']}"
csv_filename = csv_dir / f"{table_id_str.replace(' ', '_')}.csv"
df.to_csv(csv_filename, index=False)
console_warning(f"Created file: {csv_filename}")
# Only show top N kernels (as specified in --max-kernel-num)
# in "Top Stats" section
if table_type == "raw_csv_table" and table_config["source"] in [
"pmc_kernel_top.csv",
"pmc_dispatch_info.csv",
]:
df = df.head(args.max_stat_num)
# NB:
# "columnwise: True" is a special attr of a table/df
# For raw_csv_table, such as system_info, we transpose the
# df when load it, because we need those items in column.
# For metric_table, we only need to show the data in column
# fash for now.
transpose = table_type != "raw_csv_table" and table_config.get("columnwise", False)
# enable mem_chart only with single run
if (
table_config.get("cli_style") == "mem_chart"
and len(runs) == 1
and "Metric" in df.columns
and "Value" in df.columns
):
mem_data = (
pd.DataFrame([df["Metric"], df["Value"]])
.transpose()
.set_index("Metric")
.to_dict()["Value"]
)
content += mem_chart.plot_mem_chart("", args.normal_unit, mem_data) + "\n"
else:
content += (
get_table_string(df, transpose=transpose, decimal=args.decimal) + "\n"
)
return content
def show_all(
args: argparse.Namespace,
runs: dict[str, Any],
arch_configs: schema.ArchConfig,
output: Optional[TextIO],
profiling_config: dict[str, Any],
roof_plot: Optional[str] = None,
) -> None:
"""
Show all panels with their data in plain text mode.
"""
comparable_columns = parser.build_comparable_columns(args.time_unit)
filter_panel_ids = profiling_config.get("filter_blocks", [])
csv_dir = None
if isinstance(filter_panel_ids, dict):
# For backward compatibility
filter_panel_ids = [
name for name, type in filter_panel_ids.items() if type == "metric_id"
name
for name, table_type in filter_panel_ids.items()
if table_type == "metric_id"
]
filter_panel_ids = [
int(convert_metric_id_to_panel_info(metric_id)[0])
int(result[0])
for metric_id in filter_panel_ids
if (result := convert_metric_id_to_panel_info(metric_id)) is not None
]
if args.include_cols:
hidden_cols = list(set(config.HIDDEN_COLUMNS_CLI) - set(args.include_cols))
else:
@@ -149,119 +462,19 @@ def show_all(args, runs, archConfigs, output, profiling_config, roof_plot=None):
if not csv_dir.exists():
csv_dir.mkdir()
for panel_id, panel in archConfigs.panel_configs.items():
for panel_id, panel in arch_configs.panel_configs.items():
# Skip panels that don't support baseline comparison
if len(args.path) > 1 and panel_id in config.HIDDEN_SECTIONS:
continue
ss = "" # store content of all data_source from one panel
panel_content = "" # store content of all data_source from one panel
if panel_id == 400:
has_roofline_style = any(
data_source.get(type, {}).get("cli_style") == "Roofline"
for data_source in panel["data source"]
for type in data_source
)
if has_roofline_style and (
not args.filter_metrics or "4" in args.filter_metrics
):
print("\n" + "=" * 80, file=output)
print("4. Roofline", file=output)
print("=" * 80, file=output)
for run_path, workload in runs.items():
if (
hasattr(workload, "roofline_metrics")
and workload.roofline_metrics
):
print(
"\n(4.1) Per-Kernel Roofline Metrics and "
"(4.2) AI Plot Points",
file=output,
)
print("-" * 80, file=output)
kernel_top_df = workload.dfs.get(1, pd.DataFrame())
if not kernel_top_df.empty:
kernel_name_shortener(kernel_top_df, args.kernel_verbose)
for i, (kernel_id, metrics) in enumerate(
workload.roofline_metrics.items()
):
if (
not kernel_top_df.empty
and kernel_id in kernel_top_df.index
):
kernel_name = kernel_top_df.loc[
kernel_id, "Kernel_Name"
]
kernel_pct = (
kernel_top_df.loc[kernel_id, "Pct"]
if "Pct" in kernel_top_df.columns
else 0
)
else:
kernel_name = metrics.get("name", f"Kernel {kernel_id}")
kernel_pct = 0
display_name = (
kernel_name[:80] + "..."
if len(kernel_name) > 80
else kernel_name
)
print(
f"\nKernel {kernel_id}: "
f"{display_name} "
f"({kernel_pct:.1f}%)",
file=output,
)
base_indent = " "
table_indent_prefix = f"{base_indent}| "
tables = {
401: (
"4.1 Roofline Rate Metrics:",
metrics.get("ai_table", pd.DataFrame()),
),
402: (
"4.2 Roofline AI Plot Points:",
metrics.get("calc_table", pd.DataFrame()),
),
}
print(f"{base_indent}|")
for table_id, (table_name, df) in tables.items():
if df.empty:
continue
print(f"{base_indent}├─ {table_name}", file=output)
display_df = df.copy()
for col in hidden_cols:
if col in display_df.columns:
display_df = display_df.drop(columns=[col])
table_string = get_table_string(
display_df, transpose=False, decimal=args.decimal
)
indented_table_string = textwrap.indent(
table_string, table_indent_prefix
)
print(indented_table_string, file=output)
else:
print("\nNo per-kernel metrics available", file=output)
# Show the roofline plot
if roof_plot:
show_roof_plot(roof_plot)
if is_roofline_shown(args, runs, output, panel, roof_plot, hidden_cols):
continue
for data_source in panel["data source"]:
for type, table_config in data_source.items():
for table_type, table_config in data_source.items():
# If block filtering was used during analysis, then don't use profiling
# config. If block filtering was used in profiling config, only show
# those panels. If block filtering not used in profiling config, show
@@ -275,14 +488,12 @@ def show_all(args, runs, archConfigs, output, profiling_config, roof_plot=None):
and panel_id > 100
):
table_id_str = (
str(table_config["id"] // 100)
+ "."
+ str(table_config["id"] % 100)
f"{table_config['id'] // 100}.{table_config['id'] % 100}"
)
console_log(
f"Not showing table not selected during profiling: "
f"{table_id_str} "
f"{table_config['title']}"
f"{table_id_str} {table_config['title']}"
)
continue
@@ -290,273 +501,58 @@ def show_all(args, runs, archConfigs, output, profiling_config, roof_plot=None):
# We cannot guarantee that all runs have the same metrics.
# Only show common metrics.
if (
type == "metric_table"
table_type == "metric_table"
and "Metric" in table_config["header"].values()
and len(runs) > 1
):
# Common metrics across all runs
common_metrics = set()
for _, data in runs.items():
if not common_metrics:
common_metrics = set(data.dfs[table_config["id"]]["Metric"])
else:
common_metrics &= set(
data.dfs[table_config["id"]]["Metric"]
)
# Find common metrics across all runs
common_metrics: set[str] = set()
for run_data in runs.values():
run_metrics = set(run_data.dfs[table_config["id"]]["Metric"])
common_metrics = (
run_metrics
if not common_metrics
else common_metrics & run_metrics
)
# Apply common metrics across all runs
# Reindex all runs based on first run
initial_index = None
for key in runs.keys():
runs[key].dfs[table_config["id"]] = (
runs[key]
.dfs[table_config["id"]]
.loc[lambda d: d["Metric"].isin(common_metrics)]
)
for run_data in runs.values():
run_data.dfs[table_config["id"]] = run_data.dfs[
table_config["id"]
].loc[lambda df: df["Metric"].isin(common_metrics)]
if initial_index is None:
initial_index = runs[key].dfs[table_config["id"]].index
initial_index = run_data.dfs[table_config["id"]].index
else:
runs[key].dfs[table_config["id"]].index = initial_index
run_data.dfs[table_config["id"]].index = initial_index
# take the 1st run as baseline
base_run, base_data = next(iter(runs.items()))
base_df = base_data.dfs[table_config["id"]]
processed_df = process_table_data(
args,
runs,
table_config,
table_type,
comparable_columns,
hidden_cols,
)
if args.time_unit and has_time_data(base_df):
base_df = convert_time_columns(base_df, args.time_unit)
df = pd.DataFrame(index=base_df.index)
for header in list(base_df.keys()):
# For raw csv table, columns cannot be filtered
# If columns are filtered, then skip the headers not in
# filtered columns
if (
type == "raw_csv_table"
or not args.cols
or base_df.columns.get_loc(header) in args.cols
):
if header in hidden_cols:
pass
elif header not in comparable_columns:
if (
type == "raw_csv_table"
and (
table_config["source"] == "pmc_kernel_top.csv"
or table_config["source"] == "pmc_dispatch_info.csv"
)
and header == "Kernel_Name"
):
# NB: the width of kernel name might depend
# on the header of the table.
if table_config["source"] == "pmc_kernel_top.csv":
adjusted_name = base_df["Kernel_Name"].apply(
lambda x: string_multiple_lines(x, 40, 3)
)
else:
adjusted_name = base_df["Kernel_Name"].apply(
lambda x: string_multiple_lines(x, 80, 4)
)
df = pd.concat([df, adjusted_name], axis=1)
elif type == "raw_csv_table" and header == "Info":
for run, data in runs.items():
cur_df = data.dfs[table_config["id"]]
df = pd.concat([df, cur_df[header]], axis=1)
else:
df = pd.concat([df, base_df[header]], axis=1)
else:
for run, data in runs.items():
cur_df = data.dfs[table_config["id"]]
if args.time_unit and has_time_data(base_df):
cur_df = convert_time_columns(
cur_df, args.time_unit
)
if (type == "raw_csv_table") or (
type == "metric_table"
and (not header in hidden_cols)
):
if run != base_run:
# calc percentage over the baseline
base_df[header] = [
float(x) if x != "" else float(0)
for x in base_df[header]
]
cur_df[header] = [
float(x) if x != "" else float(0)
for x in cur_df[header]
]
t_df = pd.concat(
[
base_df[header],
cur_df[header],
],
axis=1,
)
absolute_diff = (
t_df.iloc[:, 1] - t_df.iloc[:, 0]
).round(args.decimal)
t_df = absolute_diff / t_df.iloc[:, 0].replace(
0, 1
)
if args.verbose >= 2:
console_log("---------", header, t_df)
t_df_pretty = (
t_df.astype(float)
.mul(100)
.round(args.decimal)
)
# show value + percentage
# TODO: better alignment
t_df = (
cur_df[header]
.astype(float)
.round(args.decimal)
.map(str)
.astype(str)
+ " ("
+ t_df_pretty.map(str)
+ "%)"
)
df = pd.concat([df, t_df], axis=1)
# DEBUG: When in a CI setting and flag is set,
# then verify metrics meet threshold
# requirement
if (
header in ["Value", "Count", "Avg"]
and t_df_pretty.abs()
.gt(args.report_diff)
.any()
):
df["Abs Diff"] = absolute_diff
if args.report_diff:
violation_idx = t_df_pretty.index[
t_df_pretty.abs() > args.report_diff
]
console_warning(
"Dataframe diff exceeds %s "
"threshold requirement\n"
"See metric %s"
% (
str(args.report_diff) + "%",
violation_idx.to_numpy(),
)
)
console_warning(df)
else:
cur_df_copy = copy.deepcopy(cur_df)
cur_df_copy[header] = [
(
round(float(x), args.decimal)
if x != ""
else x
)
for x in base_df[header]
]
df = pd.concat(
[df, cur_df_copy[header]], axis=1
)
if not df.empty:
# subtitle for each table in a panel if existing
table_id_str = (
str(table_config["id"] // 100)
+ "."
+ str(table_config["id"] % 100)
if not processed_df.empty:
panel_content += format_table_output(
args, table_config, processed_df, table_type, runs, csv_dir
)
# Check if any column in df is empty
is_empty_columns_exist = any([
df.columns[col_idx]
for col_idx in range(len(df.columns))
if df.replace("", None).iloc[:, col_idx].isnull().all()
])
# Do not print the table if any column is empty
if is_empty_columns_exist:
if "title" in table_config:
console_log(
f"Not showing table with empty column(s): "
f"{table_id_str} "
f"{table_config['title']}"
)
else:
console_log(
f"Not showing table with empty column(s): "
f"{table_id_str}"
)
if (
"title" in table_config
and table_config["title"]
and not is_empty_columns_exist
):
ss += table_id_str + " " + table_config["title"] + "\n"
if args.output_format == "csv" and csv_dir.is_dir():
if "title" in table_config and table_config["title"]:
table_id_str += "_" + table_config["title"]
csv_filename = str(
csv_dir.joinpath(table_id_str.replace(" ", "_") + ".csv"),
)
df.to_csv(csv_filename, index=False)
console_warning(f"Created file: {csv_filename}")
# Only show top N kernels (as specified in --max-kernel-num)
# in "Top Stats" section
if type == "raw_csv_table" and (
table_config["source"] == "pmc_kernel_top.csv"
or table_config["source"] == "pmc_dispatch_info.csv"
):
df = df.head(args.max_stat_num)
# NB:
# "columnwise: True" is a special attr of a table/df
# For raw_csv_table, such as system_info, we transpose the
# df when load it, because we need those items in column.
# For metric_table, we only need to show the data in column
# fash for now.
transpose = (
type != "raw_csv_table"
and "columnwise" in table_config
and table_config["columnwise"]
)
if not is_empty_columns_exist:
# enable mem_chart only with single run
if (
"cli_style" in table_config
and table_config["cli_style"] == "mem_chart"
and len(runs) == 1
):
# NB: to avoid broken test with
# arbitrary number with "--cols" option
if "Metric" in df.columns and "Value" in df.columns:
ss += mem_chart.plot_mem_chart(
"",
args.normal_unit,
pd.DataFrame([df["Metric"], df["Value"]])
.transpose()
.set_index("Metric")
.to_dict()["Value"],
)
ss += "\n"
else:
ss += (
get_table_string(
df, transpose=transpose, decimal=args.decimal
)
+ "\n"
)
if ss:
print("\n" + "-" * 80, file=output)
print(str(panel_id // 100) + ". " + panel["title"], file=output)
print(ss, file=output)
if panel_content:
print(f"\n{'-' * 80}", file=output)
print(f"{panel_id // 100}. {panel['title']}", file=output)
print(panel_content, file=output)
def show_roof_plot(roof_plot):
def show_roof_plot(roof_plot: str) -> None:
# TODO: short term solution to display roofline plot
print("\n" + "-" * 80)
print(f"\n{'-' * 80}")
print("4. Roofline")
print("4.3 Roofline Plot")
if roof_plot:
print(roof_plot)
else:
@@ -567,35 +563,47 @@ def show_roof_plot(roof_plot):
)
def show_kernel_stats(args, runs, archConfigs, output):
def show_kernel_stats(
args: argparse.Namespace,
runs: dict[str, Any],
arch_configs: schema.ArchConfig,
output: Optional[TextIO],
) -> None:
"""
Show the kernels and dispatches from "Top Stats" section.
"""
df = pd.DataFrame()
for panel_id, panel in archConfigs.panel_configs.items():
for panel_id, panel in arch_configs.panel_configs.items():
for data_source in panel["data source"]:
for type, table_config in data_source.items():
for table_type, table_config in data_source.items():
for run, data in runs.items():
df = pd.DataFrame()
single_df = data.dfs[table_config["id"]]
# NB:
# For pmc_kernel_top.csv, have to sort here if not
# sorted when load_table_data.
if table_config["id"] == 1:
print("\n" + "-" * 80, file=output)
print(f"\n{'-' * 80}", file=output)
print(
"Detected Kernels (sorted descending by duration)",
file=output,
)
df = pd.concat([df, single_df["Kernel_Name"]], axis=1)
display_df = pd.DataFrame()
display_df = pd.concat(
[display_df, single_df["Kernel_Name"]], axis=1
)
print(
get_table_string(
display_df, transpose=False, decimal=args.decimal
),
file=output,
)
if table_config["id"] == 2:
print("\n" + "-" * 80, file=output)
print(f"\n{'-' * 80}", file=output)
print("Dispatch list", file=output)
df = single_df
print(
get_table_string(df, transpose=False, decimal=args.decimal),
file=output,
)
print(
get_table_string(
single_df, transpose=False, decimal=args.decimal
),
file=output,
)
projects/rocprofiler-compute/src/utils/utils.py
+366 -402
View File
File diff suppressed because it is too large Load Diff