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

Spatial multiplexing: part2, analysis (#542)

Browse Source 
Co-authored-by: Yang Wang <ywang103@amd.com>
This commit is contained in:
ywang103-amd
2025-02-13 16:47:25 -05:00
committed by Vignesh Edithal
parent 9643afa62d
commit 808a9a79ed
7 changed files with 202 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/argparser.py
+8
View File
@@ -495,6 +495,14 @@ Examples:
nargs="+",
help="\t\tSpecify GPU id(s) for filtering.",
)
analyze_group.add_argument(
"--spatial-multiplexing",
dest="spatial_multiplexing",
required=False,
default=False,
action="store_true",
help="\t\t\tMode of spatial multiplexing.",
)
analyze_group.add_argument(
"-o",
"--output",
src/rocprof_compute_analyze/analysis_base.py
+12 -1
View File
@@ -36,6 +36,7 @@ from utils.utils import (
console_log,
demarcate,
is_workload_empty,
merge_counters_spatial_multiplex,
)
@@ -57,6 +58,10 @@ class OmniAnalyze_Base:
def get_socs(self):
return self.__socs
@demarcate
def spatial_multiplex_merge_counters(self, df):
return merge_counters_spatial_multiplex(df)
@demarcate
def generate_configs(self, arch, config_dir, list_stats, filter_metrics, sys_info):
single_panel_config = file_io.is_single_panel_config(
@@ -142,6 +147,7 @@ class OmniAnalyze_Base:
sysinfo_path = (
Path(d[0])
if self.__args.nodes is None
and self.__args.spatial_multiplexing is not True
else file_io.find_1st_sub_dir(d[0])
)
sys_info = file_io.load_sys_info(sysinfo_path.joinpath("sysinfo.csv"))
@@ -166,6 +172,7 @@ class OmniAnalyze_Base:
sysinfo_path = (
Path(d[0])
if self.__args.nodes is None
and self.__args.spatial_multiplexing is not True
else file_io.find_1st_sub_dir(d[0])
)
w.sys_info = file_io.load_sys_info(sysinfo_path.joinpath("sysinfo.csv"))
@@ -199,7 +206,11 @@ class OmniAnalyze_Base:
# validate profiling data
# Todo: more err check
if not (self.__args.nodes != None or self.__args.list_nodes):
if not (
self.__args.nodes != None
or self.__args.list_nodes
or self.__args.spatial_multiplexing
):
is_workload_empty(dir[0])
# else:
src/rocprof_compute_analyze/analysis_cli.py
+6
View File
@@ -44,10 +44,16 @@ class cli_analysis(OmniAnalyze_Base):
self._runs[d[0]].raw_pmc = file_io.create_df_pmc(
d[0],
self.get_args().nodes,
self.get_args().spatial_multiplexing,
self.get_args().kernel_verbose,
self.get_args().verbose,
)
if self.get_args().spatial_multiplexing:
self._runs[d[0]].raw_pmc = self.spatial_multiplex_merge_counters(
self._runs[d[0]].raw_pmc
)
file_io.create_df_kernel_top_stats(
df_in=self._runs[d[0]].raw_pmc,
raw_data_dir=d[0],
src/rocprof_compute_analyze/analysis_webui.py
+13
View File
@@ -113,9 +113,16 @@ class webui_analysis(OmniAnalyze_Base):
base_data[base_run].raw_pmc = file_io.create_df_pmc(
self.dest_dir,
self.get_args().nodes,
self.get_args().spatial_multiplexing,
self.get_args().kernel_verbose,
self.get_args().verbose,
)
if self.get_args().spatial_multiplexing:
base_data[base_run].raw_pmc = self.spatial_multiplex_merge_counters(
base_data[base_run].raw_pmc
)
console_debug("analysis", "gui dispatch filter is %s" % disp_filt)
console_debug("analysis", "gui kernel filter is %s" % kernel_filter)
console_debug("analysis", "gui gpu filter is %s" % gcd_filter)
@@ -290,6 +297,12 @@ class webui_analysis(OmniAnalyze_Base):
self.get_args().kernel_verbose,
args.verbose,
)
if self.get_args().spatial_multiplexing:
self._runs[self.dest_dir].raw_pmc = self.spatial_multiplex_merge_counters(
self._runs[self.dest_dir].raw_pmc
)
file_io.create_df_kernel_top_stats(
df_in=self._runs[self.dest_dir].raw_pmc,
raw_data_dir=self.dest_dir,
src/rocprof_compute_base.py
+10
View File
@@ -219,6 +219,15 @@ class RocProfCompute:
p.mkdir(parents=True, exist_ok=False)
except FileExistsError:
console_error("Directory already exists.")
elif self.__args.mode == "analyze":
# block all filters during spatial-multiplexing
if self.__args.spatial_multiplexing:
self.__args.gpu_id = None
self.__args.gpu_kernel = None
self.__args.gpu_dispatch_id = None
self.__args.nodes = None
return
@demarcate
@@ -342,6 +351,7 @@ class RocProfCompute:
sysinfo_path = (
Path(d[0])
if analyzer.get_args().nodes is None
and analyzer.get_args().spatial_multiplexing is not True
else file_io.find_1st_sub_dir(d[0])
)
sys_info = file_io.load_sys_info(sysinfo_path.joinpath("sysinfo.csv"))
src/utils/file_io.py
+31 -16
View File
@@ -176,7 +176,9 @@ def create_df_kernel_top_stats(
@demarcate
def create_df_pmc(raw_data_root_dir, nodes, kernel_verbose, verbose):
def create_df_pmc(
raw_data_root_dir, nodes, spatial_multiplexing, kernel_verbose, verbose
):
"""
Load all raw pmc counters and join into one df.
"""
@@ -212,12 +214,7 @@ def create_df_pmc(raw_data_root_dir, nodes, kernel_verbose, verbose):
console_debug("pmc_raw_data final_single_df %s" % final_df.info)
return final_df
# regular single node case
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:
if spatial_multiplexing:
df = pd.DataFrame()
# todo: more err check
for subdir in Path(raw_data_root_dir).iterdir():
@@ -230,15 +227,33 @@ def create_df_pmc(raw_data_root_dir, nodes, kernel_verbose, verbose):
# specified node list
else:
df = pd.DataFrame()
# todo: more err check
for subdir in nodes:
p = Path(raw_data_root_dir)
new_df = create_single_df_pmc(
p.joinpath(subdir), subdir, kernel_verbose, verbose
)
df = pd.concat([df, new_df])
return df
# regular single node case
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
# specified node list
else:
df = pd.DataFrame()
# todo: more err check
for subdir in nodes:
p = Path(raw_data_root_dir)
new_df = create_single_df_pmc(
p.joinpath(subdir), subdir, kernel_verbose, verbose
)
df = pd.concat([df, new_df])
return df
def collect_wave_occu_per_cu(in_dir, out_dir, numSE):
src/utils/utils.py
+122
View File
@@ -1091,3 +1091,125 @@ def set_locale_encoding():
exit=False,
)
console_error(error)
def reverse_multi_index_df_pmc(final_df):
"""
Util function to decompose multi-index dataframe.
"""
# Check if the columns have more than one level
if len(final_df.columns.levels) < 2:
raise ValueError("Input DataFrame does not have a multi-index column.")
# Extract the first level of the MultiIndex columns (the file names)
coll_levels = final_df.columns.get_level_values(0).unique().tolist()
# Initialize the list of DataFrames
dfs = []
# Loop through each 'coll_level' and rebuild the DataFrames
for level in coll_levels:
# Select columns that belong to the current 'coll_level'
columns_for_level = final_df.xs(level, axis=1, level=0)
# Append the DataFrame for this level
dfs.append(columns_for_level)
# Return the list of DataFrames and the column levels
return dfs, coll_levels
def merge_counters_spatial_multiplex(df_multi_index):
"""
For spatial multiplexing, this merges counter values for the same kernel that runs on different devices. For time stamp, start time stamp will use median while for end time stamp, it will be equal to the summation between median start stamp and median delta time.
"""
non_counter_column_index = [
"Dispatch_ID",
"GPU_ID",
"Queue_ID",
"PID",
"TID",
"Grid_Size",
"Workgroup_Size",
"LDS_Per_Workgroup",
"Scratch_Per_Workitem",
"Arch_VGPR",
"Accum_VGPR",
"SGPR",
"Wave_Size",
"Kernel_Name",
"Start_Timestamp",
"End_Timestamp",
"Correlation_ID",
"Kernel_ID",
"Node",
]
expired_column_index = [
"Node",
"PID",
"TID",
"Queue_ID",
]
result_dfs = []
# TODO: will need optimize to avoid this convertion to single index format and do merge directly on multi-index dataframe
dfs, coll_levels = reverse_multi_index_df_pmc(df_multi_index)
for df in dfs:
kernel_name_column_name = "Kernel_Name"
if not "Kernel_Name" in df and "Name" in df:
kernel_name_column_name = "Name"
# Find the values in Kernel_Name that occur more than once
kernel_single_occurances = df[kernel_name_column_name].value_counts().index
# Define a list to store the merged rows
result_data = []
for kernel_name in kernel_single_occurances:
# Get all rows for the current kernel_name
group = df[df[kernel_name_column_name] == kernel_name]
# Create a dictionary to store the merged row for the current group
merged_row = {}
# Process non-counter columns
for col in [
col for col in non_counter_column_index if col not in expired_column_index
]:
if col == "Start_Timestamp":
# For Start_Timestamp, take the median
merged_row[col] = group["Start_Timestamp"].median()
elif col == "End_Timestamp":
# For End_Timestamp, calculate the median delta time
delta_time = group["End_Timestamp"] - group["Start_Timestamp"]
median_delta_time = delta_time.median()
merged_row[col] = merged_row["Start_Timestamp"] + median_delta_time
else:
# For other non-counter columns, take the first occurrence (0th row)
merged_row[col] = group.iloc[0][col]
# Process counter columns (assumed to be all columns not in non_counter_column_index)
counter_columns = [
col for col in group.columns if col not in non_counter_column_index
]
for counter_col in counter_columns:
# for counter columns, take the first non-none (or non-nan) value
current_valid_counter_group = group[group[counter_col].notna()]
first_valid_value = (
current_valid_counter_group.iloc[0][counter_col]
if len(current_valid_counter_group) > 0
else None
)
merged_row[counter_col] = first_valid_value
# Append the merged row to the result list
result_data.append(merged_row)
# Create a new DataFrame from the merged rows
result_dfs.append(pd.DataFrame(result_data))
final_df = pd.concat(result_dfs, keys=coll_levels, axis=1, copy=False)
return final_df