Enable rocpd output format with rocprofiler sdk (#790)

* Add `rocpd` choice for `--format-rocprof-output` option
* Add rocpd_data.py which defines SQL queries to extract data from rocpd database
* Use sqlite3 package to read the database
* Add `--retain-rocpd-output` option in profile mode to retain raw
  rocpd database
* Add warning notice to say `--format-rocprof-output rocpd` will be
  default in future release

For rocpd output:
* Use only `pmc_perf.csv` instead of reading individual coll_level results csv files
* Post process csv files using pandas in analysis mode instead of profile mode
* Use ACCUM counters instead of SQ_ACCUM_PREV_HIRES

* Add test cases for rocpd output format
* Fix code formatting issues
* Update CHANGELOG

[ROCm/rocprofiler-compute commit: 03d27c0ba0]

projects/rocprofiler-compute/src/argparser.py

@@ -249,7 +249,7 @@ Examples:
         required=False,
         metavar="",
         dest="format_rocprof_output",
-        choices=["json", "csv"],
+        choices=["json", "csv", "rocpd"],
         default="csv",
         help="\t\t\tSet the format of output file of rocprof.",
     )
@@ -280,6 +280,13 @@ Examples:
         default="/opt/rocm/lib/librocprofiler-sdk.so",
         help="\t\t\tSet the path to rocprofiler SDK library.",
     )
+    profile_group.add_argument(
+        "--retain-rocpd-output",
+        required=False,
+        default=False,
+        action="store_true",
+        help="\t\t\tRetain the large raw rocpd database in workload directory.\n\t\t\tThis option requires --format-rocprof-output rocpd.",
+    )
 
     ## Roofline Command Line Options
     roofline_group.add_argument(

projects/rocprofiler-compute/src/rocprof_compute_analyze/analysis_base.py

@@ -128,11 +128,13 @@ class OmniAnalyze_Base:
         if not normalization_filter:
             for k, v in self._arch_configs.items():
                 parser.build_metric_value_string(
-                    v.dfs, v.dfs_type, self.__args.normal_unit
+                    v.dfs, v.dfs_type, self.__args.normal_unit, self._profiling_config
                 )
         else:
             for k, v in self._arch_configs.items():
-                parser.build_metric_value_string(v.dfs, v.dfs_type, normalization_filter)
+                parser.build_metric_value_string(
+                    v.dfs, v.dfs_type, normalization_filter, self._profiling_config
+                )
 
         args = self.__args
         # Error checking for multiple runs and multiple kernel filters

projects/rocprofiler-compute/src/rocprof_compute_analyze/analysis_cli.py

@@ -47,6 +47,7 @@ class cli_analysis(OmniAnalyze_Base):
                 self.get_args().spatial_multiplexing,
                 self.get_args().kernel_verbose,
                 self.get_args().verbose,
+                self._profiling_config,
             )
 
             if self.get_args().spatial_multiplexing:
@@ -72,7 +73,11 @@ class cli_analysis(OmniAnalyze_Base):
 
             # create the loaded table
             parser.load_table_data(
-                workload=self._runs[d[0]], dir=d[0], is_gui=False, args=self.get_args()
+                workload=self._runs[d[0]],
+                dir=d[0],
+                is_gui=False,
+                args=self.get_args(),
+                config=self._profiling_config,
             )
 
     @demarcate

projects/rocprofiler-compute/src/rocprof_compute_analyze/analysis_webui.py

@@ -120,6 +120,7 @@ class webui_analysis(OmniAnalyze_Base):
                 self.get_args().spatial_multiplexing,
                 self.get_args().kernel_verbose,
                 self.get_args().verbose,
+                self._profiling_config,
             )
 
             if self.get_args().spatial_multiplexing:
@@ -168,6 +169,7 @@ class webui_analysis(OmniAnalyze_Base):
                 dir=self.dest_dir,
                 is_gui=True,
                 args=self.get_args(),
+                config=self._profiling_config,
             )
 
             # ~~~~~~~~~~~~~~~~~~~~~~~
@@ -300,6 +302,7 @@ class webui_analysis(OmniAnalyze_Base):
                 self.get_args().spatial_multiplexing,
                 self.get_args().kernel_verbose,
                 args.verbose,
+                self._profiling_config,
             )
 
             if self.get_args().spatial_multiplexing:

projects/rocprofiler-compute/src/rocprof_compute_base.py

@@ -169,6 +169,14 @@ class RocProfCompute:
         )
         self.__args = parser.parse_args()
 
+        if (
+            "format_rocprof_output" in self.__args
+            and self.__args.format_rocprof_output != "rocpd"
+        ):
+            console_warning(
+                f"The option --format-rocprof-output currently set to {self.__args.format_rocprof_output} will default to rocpd in a future release."
+            )
+
         if self.__args.mode == None:
             if self.__args.specs:
                 print(generate_machine_specs(self.__args))

projects/rocprofiler-compute/src/rocprof_compute_profile/profiler_base.py

@@ -22,6 +22,7 @@
 # SOFTWARE.
 ##############################################################################el
 
+import csv
 import glob
 import logging
 import os
@@ -68,6 +69,30 @@ class RocProfCompute_Base:
     @demarcate
     def join_prof(self, out=None):
         """Manually join separated rocprof runs"""
+        if self.get_args().format_rocprof_output == "rocpd":
+            # Vertically concat (by rows) results_*.csv into pmc_perf.csv
+            result_files = glob.glob(self.get_args().path + "/results_*.csv")
+            if out is None:
+                out = self.__args.path + "/pmc_perf.csv"
+            with open(out, "w", newline="") as outfile:
+                writer = None
+                for file in result_files:
+                    with open(file, "r", newline="") as infile:
+                        reader = csv.reader(infile)
+                        header = next(reader)
+                        # Write header only once
+                        if writer is None:
+                            writer = csv.writer(outfile)
+                            writer.writerow(header)
+                        for row in reader:
+                            writer.writerow(row)
+            console_debug(f"Created file: {out}")
+            # Delete results_*.csv files
+            for file in result_files:
+                os.remove(file)
+                console_debug(f"Deleted file: {file}")
+            return
+
         # Set default output directory if not specified
         if type(self.__args.path) == str:
             if out is None:
@@ -412,6 +437,7 @@ class RocProfCompute_Base:
                     mspec=self._soc._mspec,
                     loglevel=self.get_args().loglevel,
                     format_rocprof_output=self.get_args().format_rocprof_output,
+                    retain_rocpd_output=self.get_args().retain_rocpd_output,
                 )
                 end_run_prof = time.time()
                 actual_profiling_duration = end_run_prof - start_run_prof

projects/rocprofiler-compute/src/rocprof_compute_profile/profiler_rocprof_v3.py

@@ -43,11 +43,6 @@ class rocprof_v3_profiler(RocProfCompute_Base):
     def get_profiler_options(self, fname, soc):
         app_cmd = shlex.split(self.get_args().remaining)
         trace_option = "--kernel-trace"
-        rocprof_out_format = "json"
-
-        if self.get_args().format_rocprof_output == "csv":
-            rocprof_out_format = "csv"
-
         if self.get_args().kokkos_trace:
             trace_option = "--kokkos-trace"
             # NOTE: --kokkos-trace feature is incomplete and is disabled for now.
@@ -63,7 +58,7 @@ class rocprof_v3_profiler(RocProfCompute_Base):
             self.get_args().path + "/" + "out",
             trace_option,
             "--output-format",
-            rocprof_out_format,
+            self.get_args().format_rocprof_output,
         ]
         # Kernel filtering
         if self.get_args().kernel:

projects/rocprofiler-compute/src/rocprof_compute_profile/profiler_rocprofiler_sdk.py

@@ -55,13 +55,10 @@ class rocprofiler_sdk_profiler(RocProfCompute_Base):
             "ROCP_TOOL_LIBRARIES": rocprofiler_sdk_tool_path,
             "LD_LIBRARY_PATH": rocm_libdir,
             "ROCPROF_KERNEL_TRACE": "1",
-            "ROCPROF_OUTPUT_FORMAT": "json",
+            "ROCPROF_OUTPUT_FORMAT": self.get_args().format_rocprof_output,
             "ROCPROF_OUTPUT_PATH": self.get_args().path + "/out/pmc_1",
         }
 
-        if self.get_args().format_rocprof_output == "csv":
-            options["ROCPROF_OUTPUT_FORMAT"] = "csv"
-
         if self.get_args().kokkos_trace:
             # NOTE: --kokkos-trace feature is incomplete and is disabled for now.
             console_error(

projects/rocprofiler-compute/src/rocprof_compute_tui/analysis_tui.py

@@ -61,6 +61,7 @@ class tui_analysis(OmniAnalyze_Base):
             self.get_args().spatial_multiplexing,
             self.get_args().kernel_verbose,
             self.get_args().verbose,
+            self._profiling_config,
         )
 
         if self.get_args().spatial_multiplexing:
@@ -90,6 +91,7 @@ class tui_analysis(OmniAnalyze_Base):
             dir=self.path,
             is_gui=False,
             args=self.get_args(),
+            config=self._profiling_config,
         )
 
     def initalize_runs(self, normalization_filter=None):

projects/rocprofiler-compute/src/roofline.py

@@ -35,6 +35,7 @@ import plotext as plt
 import plotly.graph_objects as go
 from dash import dcc, html
 
+from utils import file_io, rocpd_data
 from utils.logger import (
     console_debug,
     console_error,
@@ -673,6 +674,9 @@ class Roofline:
             console_error("roofline", "{} does not exist".format(pmc_perf_csv))
         t_df = OrderedDict()
         t_df["pmc_perf"] = pd.read_csv(pmc_perf_csv)
+        profiling_config = file_io.load_profiling_config(self.__args.path[0][0])
+        if profiling_config.get("format_rocprof_output") == "rocpd":
+            t_df["pmc_perf"] = rocpd_data.process_rocpd_csv(t_df["pmc_perf"])
 
         color_scheme = {
             "HBM": "blue+",
@@ -861,6 +865,9 @@ class Roofline:
             console_error("roofline", "{} does not exist".format(app_path))
         t_df = OrderedDict()
         t_df["pmc_perf"] = pd.read_csv(app_path)
+        profiling_config = file_io.load_profiling_config(self.__args.path)
+        if profiling_config.get("format_rocprof_output") == "rocpd":
+            t_df["pmc_perf"] = rocpd_data.process_rocpd_csv(t_df["pmc_perf"])
         self.empirical_roofline(ret_df=t_df)
 
     @abstractmethod

projects/rocprofiler-compute/src/utils/file_io.py

@@ -31,7 +31,7 @@ import pandas as pd
 import yaml
 
 import config
-from utils import schema
+from utils import rocpd_data, schema
 from utils.kernel_name_shortener import kernel_name_shortener
 from utils.logger import console_debug, console_error, console_log, demarcate
 
@@ -95,9 +95,7 @@ def load_profiling_config(config_dir):
             prof_config = yaml.safe_load(file)
             return prof_config
     except FileNotFoundError:
-        console_log(
-            f"Could not find profiling_config.yaml in {config_dir} for filtering analysis report"
-        )
+        console_log(f"Could not find profiling_config.yaml in {config_dir}")
     return dict()
 
 
@@ -195,7 +193,7 @@ def create_df_kernel_top_stats(
 
 @demarcate
 def create_df_pmc(
-    raw_data_root_dir, nodes, spatial_multiplexing, kernel_verbose, verbose
+    raw_data_root_dir, nodes, spatial_multiplexing, kernel_verbose, verbose, config
 ):
     """
     Load all raw pmc counters and join into one df.
@@ -214,6 +212,8 @@ def create_df_pmc(
                     f == schema.pmc_perf_file_prefix + ".csv"
                 ):
                     tmp_df = pd.read_csv(str(Path(root).joinpath(f)))
+                    if config.get("format_rocprof_output") == "rocpd":
+                        tmp_df = rocpd_data.process_rocpd_csv(tmp_df)
                     # Demangle original KernelNames
                     kernel_name_shortener(tmp_df, kernel_verbose)
 

projects/rocprofiler-compute/src/utils/parser.py

@@ -271,7 +271,7 @@ class CodeTransformer(ast.NodeTransformer):
         return node
 
 
-def build_eval_string(equation, coll_level):
+def build_eval_string(equation, coll_level, config):
     """
     Convert user defined equation string to eval executable string
     For example,
@@ -314,7 +314,14 @@ def build_eval_string(equation, coll_level):
     # use .get() to catch any potential KeyErrors
     s = re.sub(r"raw_pmc_df\['(.*?)']", r'raw_pmc_df.get("\1")', s)
     # apply coll_level
-    s = re.sub(r"raw_pmc_df", "raw_pmc_df.get('" + coll_level + "')", s)
+    if config.get("format_rocprof_output") == "rocpd":
+        # Replace SQ_ACCUM_PREV_HIRES with coll_level_ACCUM then ignore coll_level df
+        s = re.sub(f"SQ_ACCUM_PREV_HIRES", f"{coll_level}_ACCUM", s)
+        s = re.sub(
+            r"raw_pmc_df", "raw_pmc_df.get('" + schema.pmc_perf_file_prefix + "')", s
+        )
+    else:
+        s = re.sub(r"raw_pmc_df", "raw_pmc_df.get('" + coll_level + "')", s)
     # print("--- build_eval_string, return: ", s)
     return s
 
@@ -653,7 +660,7 @@ def build_dfs(archConfigs, filter_metrics, sys_info):
     setattr(archConfigs, "metric_counters", metric_counters)
 
 
-def build_metric_value_string(dfs, dfs_type, normal_unit):
+def build_metric_value_string(dfs, dfs_type, normal_unit, profiling_config):
     """
     Apply the real eval string to its field in the metric_table df.
     """
@@ -674,6 +681,7 @@ def build_metric_value_string(dfs, dfs_type, normal_unit):
                                 df.at[row_idx_label, expr] = build_eval_string(
                                     df.at[row_idx_label, expr],
                                     df.at[row_idx_label, "coll_level"],
+                                    profiling_config,
                                 )
 
                 elif expr.lower() == "unit" or expr.lower() == "units":
@@ -683,7 +691,7 @@ def build_metric_value_string(dfs, dfs_type, normal_unit):
 
 
 @demarcate
-def eval_metric(dfs, dfs_type, sys_info, raw_pmc_df, debug):
+def eval_metric(dfs, dfs_type, sys_info, raw_pmc_df, debug, config):
     """
     Execute the expr string for each metric in the df.
     """
@@ -784,7 +792,7 @@ def eval_metric(dfs, dfs_type, sys_info, raw_pmc_df, debug):
         if "PER_XCD" not in key:
             continue
         # NB: assume all built-in vars from pmc_perf.csv for now
-        s = build_eval_string(value, schema.pmc_perf_file_prefix)
+        s = build_eval_string(value, schema.pmc_perf_file_prefix, config)
         try:
             ammolite__build_in[key] = eval(compile(s, "<string>", "eval"))
         except TypeError:
@@ -801,7 +809,7 @@ def eval_metric(dfs, dfs_type, sys_info, raw_pmc_df, debug):
         if "PER_XCD" in key:
             continue
         # NB: assume all built-in vars from pmc_perf.csv for now
-        s = build_eval_string(value, schema.pmc_perf_file_prefix)
+        s = build_eval_string(value, schema.pmc_perf_file_prefix, config)
         try:
             ammolite__build_in[key] = eval(compile(s, "<string>", "eval"))
         except TypeError:
@@ -1437,7 +1445,7 @@ def load_kernel_top(workload, dir, args):
 
 
 @demarcate
-def load_table_data(workload, dir, is_gui, args, skipKernelTop=False):
+def load_table_data(workload, dir, is_gui, args, config, skipKernelTop=False):
     """
     - Load data for all "raw_csv_table"
     - Load dat for "pc_sampling_table"
@@ -1452,6 +1460,7 @@ def load_table_data(workload, dir, is_gui, args, skipKernelTop=False):
         workload.sys_info.iloc[0],
         apply_filters(workload, dir, is_gui, args.debug),
         args.debug,
+        config,
     )
 
 

projects/rocprofiler-compute/src/utils/rocpd_data.py

@@ -0,0 +1,94 @@
+import csv
+import sqlite3
+from contextlib import closing
+
+from utils.logger import console_error
+
+# From schema definition in source/share/rocprofiler-sdk-rocpd/data_views.sql in rocprofiler-sdk repository
+COUNTERS_COLLECTION_QUERY = """
+SELECT
+    agent_id as GPU_ID,
+    dispatch_id as Dispatch_ID,
+    grid_size as Grid_Size,
+    workgroup_size as Workgroup_Size,
+    lds_block_size as LDS_Per_Workgroup,
+    scratch_size as Scratch_Per_Workitem,
+    vgpr_count as Arch_VGPR,
+    accum_vgpr_count as Accum_VGPR,
+    sgpr_count as SGPR,
+    kernel_name as Kernel_Name,
+    start as Start_Timestamp,
+    end as End_Timestamp,
+    kernel_id as Kernel_ID,
+    counter_name as Counter_Name,
+    value as Counter_Value
+FROM counters_collection
+"""
+
+
+def convert_db_to_csv(
+    db_path: str,
+    csv_file_path: str,
+) -> None:
+    """
+    Read rocpd database and write to CSV file
+    """
+    # Read counters_collection view from the database and write to CSV
+    try:
+        with closing(sqlite3.connect(db_path)) as conn:
+            with closing(conn.execute(COUNTERS_COLLECTION_QUERY)) as cursor:
+                with open(csv_file_path, "w", newline="") as csvfile:
+                    writer = csv.writer(csvfile)
+                    writer.writerow(
+                        [description[0] for description in cursor.description]
+                    )
+                    for row in cursor:
+                        writer.writerow(row)
+    except (sqlite3.DatabaseError, IOError) as e:
+        console_error(f"Error converting database to CSV: {e}")
+
+
+def process_rocpd_csv(df):
+    """
+    Merge counters across unique dispatches from the input dataframe and return processed dataframe.
+    """
+    # Only import pandas if needed
+    import pandas as pd
+
+    data = list()
+    # Group by unique kernel and merge into a single row
+    for _, group_df in df.groupby(
+        [
+            "Dispatch_ID",
+            "Kernel_Name",
+            "Grid_Size",
+            "Workgroup_Size",
+            "LDS_Per_Workgroup",
+        ]
+    ):
+        row = {
+            "GPU_ID": group_df["GPU_ID"].iloc[0],
+            "Grid_Size": group_df["Grid_Size"].iloc[0],
+            "Workgroup_Size": group_df["Workgroup_Size"].iloc[0],
+            "LDS_Per_Workgroup": group_df["LDS_Per_Workgroup"].iloc[0],
+            "Scratch_Per_Workitem": group_df["Scratch_Per_Workitem"].iloc[0],
+            "Arch_VGPR": group_df["Arch_VGPR"].iloc[0],
+            "Accum_VGPR": group_df["Accum_VGPR"].iloc[0],
+            "SGPR": group_df["SGPR"].iloc[0],
+            "Kernel_Name": group_df["Kernel_Name"].iloc[0],
+            "Kernel_ID": group_df["Kernel_ID"].iloc[0],
+        }
+        # Each counter will become its own column
+        row.update(dict(zip(group_df["Counter_Name"], group_df["Counter_Value"])))
+        # Replace end timestamp with median of durations of group, start timestamp is set to 0
+        row["End_Timestamp"] = (
+            group_df["End_Timestamp"] - group_df["Start_Timestamp"]
+        ).median()
+        row["Start_Timestamp"] = 0.0
+        data.append(row)
+    df = pd.DataFrame(data)
+    # Rank GPU IDs, map lowest number to 0, next to 1, etc.
+    df["GPU_ID"] = df["GPU_ID"].rank(method="dense").astype(int) - 1
+    # Reset dispatch IDs
+    df["Dispatch_ID"] = range(len(df))
+    return df

projects/rocprofiler-compute/src/utils/utils.py

@@ -45,6 +45,7 @@ import pandas as pd
 import yaml
 
 import config
+from utils import rocpd_data
 from utils.logger import (
     console_debug,
     console_error,
@@ -707,9 +708,14 @@ def parse_text(text_file):
 
 
 def run_prof(
-    fname, profiler_options, workload_dir, mspec, loglevel, format_rocprof_output
+    fname,
+    profiler_options,
+    workload_dir,
+    mspec,
+    loglevel,
+    format_rocprof_output,
+    retain_rocpd_output=False,
 ):
-    time_0 = time.time()
     fbase = path(fname).stem
 
     console_debug("pmc file: %s" % path(fname).name)
@@ -831,7 +837,29 @@ def run_prof(
 
     results_files = []
 
-    if rocprof_cmd.endswith("v2"):
+    if format_rocprof_output == "rocpd":
+        if rocprof_cmd == "rocprofiler-sdk" or rocprof_cmd.endswith("v3"):
+            # Write results_fbase.csv
+            rocpd_data.convert_db_to_csv(
+                glob.glob(workload_dir + "/out/pmc_1/*/*.db")[0],
+                workload_dir + f"/results_{fbase}.csv",
+            )
+            if retain_rocpd_output:
+                shutil.copyfile(
+                    glob.glob(workload_dir + "/out/pmc_1/*/*.db")[0],
+                    workload_dir + "/" + fbase + ".db",
+                )
+                console_warning(
+                    f"Retaining large raw rocpd database: {workload_dir}/{fbase}.db"
+                )
+            # Remove temp directory
+            shutil.rmtree(workload_dir + "/" + "out")
+            return
+        else:
+            console_error(
+                "rocpd output format is only supported with rocprofiler-sdk or rocprofv3."
+            )
+    elif rocprof_cmd.endswith("v2"):
         # rocprofv2 has separate csv files for each process
         results_files = glob.glob(workload_dir + "/out/pmc_1/results_*.csv")
 
@@ -1058,7 +1086,6 @@ def process_rocprofv3_output(rocprof_output, workload_dir, is_timestamps):
         else:
             # when the input is not for timestamps, and counter csv file is not generated, we assume failed rocprof run and will completely bypass the file generation and merging for current pmc
             results_files_csv = []
-
     else:
         console_error("The output file of rocprofv3 can only support json or csv!!!")