Trace: construct a CCT

pipit/readers/hpctoolkit_reader.py

@@ -1356,5 +1356,7 @@ def read(self) -> pipit.trace.Trace:
             }
         )
 
+        # cct is needed to create trace in hpctoolkit,
+        # so always return it as part of the trace
         self.trace_df = trace_df
-        return pipit.trace.Trace(None, trace_df)
+        return pipit.trace.Trace(None, trace_df, self.cct)

pipit/readers/nsight_reader.py

@@ -10,9 +10,10 @@
 class NsightReader:
     """Reader for Nsight trace files"""
 
-    def __init__(self, file_name) -> None:
+    def __init__(self, file_name, create_cct=True) -> None:
         self.file_name = file_name
         self.df = None
+        self.create_cct = create_cct
 
     def read(self):
         """
@@ -103,4 +104,8 @@ def read(self):
         # Applying the column list to the dataframe to rearrange
         self.df = self.df.loc[:, cols]
 
-        return pipit.trace.Trace(None, self.df)
+        trace = pipit.trace.Trace(None, self.df)
+        if self.create_cct:
+            trace.create_cct()
+
+        return trace

pipit/readers/otf2_reader.py

@@ -13,9 +13,10 @@
 class OTF2Reader:
     """Reader for OTF2 trace files"""
 
-    def __init__(self, dir_name, num_processes=None):
+    def __init__(self, dir_name, num_processes=None, create_cct=True):
         self.dir_name = dir_name  # directory of otf2 file being read
         self.file_name = self.dir_name + "/traces.otf2"
+        self.create_cct = create_cct
 
         num_cpus = mp.cpu_count()
         if num_processes is None or num_processes < 1 or num_processes > num_cpus:
@@ -516,4 +517,8 @@ def read(self):
 
         self.events = self.read_events()  # events
 
-        return pipit.trace.Trace(self.definitions, self.events)
+        trace = pipit.trace.Trace(self.definitions, self.events)
+        if self.create_cct:
+            trace.create_cct()
+
+        return trace

pipit/readers/projections_reader.py

@@ -3,7 +3,6 @@
 #
 # SPDX-License-Identifier: MIT
 
-
 import os
 import gzip
 import pipit.trace
@@ -225,7 +224,7 @@ def read_sts_file(self):
 
 
 class ProjectionsReader:
-    def __init__(self, projections_directory: str, num_processes=None) -> None:
+    def __init__(self, projections_directory: str, num_processes=None, create_cct=True) -> None:
         if not os.path.isdir(projections_directory):
             raise ValueError("Not a valid directory.")
 
@@ -270,6 +269,8 @@ def __init__(self, projections_directory: str, num_processes=None) -> None:
         else:
             self.num_processes = num_processes
 
+        self.create_cct = create_cct
+
     # Returns an empty dict, used for reading log file into dataframe
     @staticmethod
     def _create_empty_dict() -> dict:
@@ -317,7 +318,11 @@ def read(self):
             ["Timestamp (ns)", "Event Type", "Name", "Process", "Attributes"]
         ]
 
-        return pipit.trace.Trace(None, trace_df)
+        trace = pipit.trace.Trace(None, trace_df)
+        if self.create_cct:
+            trace.create_cct()
+
+        return trace
 
     def _read_log_file(self, rank_size) -> pd.DataFrame:
         # has information needed in sts file

pipit/trace.py

@@ -5,17 +5,20 @@
 
 import numpy as np
 import pandas as pd
+from pipit.util.cct import create_cct
 
 
 class Trace:
-    """A trace dataset is read into an object of this type, which includes one
-    or more dataframes.
+    """
+    A trace dataset is read into an object of this type, which
+    includes one or more dataframes and a calling context tree.
     """
 
-    def __init__(self, definitions, events):
+    def __init__(self, definitions, events, cct=None):
         """Create a new Trace object."""
         self.definitions = definitions
         self.events = events
+        self.cct = cct
 
         # list of numeric columns which we can calculate inc/exc metrics with
         self.numeric_cols = list(
@@ -26,6 +29,11 @@ def __init__(self, definitions, events):
         self.inc_metrics = []
         self.exc_metrics = []
 
+    def create_cct(self):
+        # adds a column of cct nodes to the events dataframe
+        # and stores the graph object in self.cct
+        self.cct = create_cct(self.events)
+
     @staticmethod
     def from_otf2(dirname, num_processes=None):
         """Read an OTF2 trace into a new Trace object."""
@@ -354,7 +362,6 @@ def comm_matrix(self, output="size"):
         Communication Matrix for Peer-to-Peer (P2P) MPI messages
 
         Arguments:
-
         1) output -
         string to choose whether the communication volume should be measured
         by bytes transferred between two processes or the number of messages

pipit/util/__init__.py

@@ -0,0 +1,4 @@
+# Copyright 2022-2023 Parallel Software and Systems Group, University of
+# Maryland. See the top-level LICENSE file for details.
+#
+# SPDX-License-Identifier: MIT

pipit/util/cct.py

@@ -0,0 +1,111 @@
+# Copyright 2022-2023 Parallel Software and Systems Group, University of
+# Maryland.  See the top-level LICENSE file for details.
+#
+# SPDX-License-Identifier: MIT
+
+from pipit.graph import Graph, Node
+
+
+def create_cct(events):
+    """
+    Generic function to iterate through the events dataframe and create a CCT.
+    Uses pipit's graph data structure for this. Returns a CCT
+    and creates a new column in the Events DataFrame that stores
+    a reference to each row's corresponding node in the CCT.
+    """
+
+    # CCT and list of nodes in DataFrame
+    graph = Graph()
+    graph_nodes = [None for i in range(len(events))]
+
+    # determines whether a node exists or not
+    callpath_to_node = dict()
+
+    node_id = 0  # each node has a unique id
+
+    # Filter the DataFrame to only Enter/Leave
+    enter_leave_df = events.loc[events["Event Type"].isin(["Enter", "Leave"])]
+
+    # list of processes and/or threads to iterate over
+    if "Thread" in events.columns:
+        exec_locations = set(zip(events["Process"], events["Thread"]))
+        has_thread = True
+    else:
+        exec_locations = set(events["Process"])
+        has_thread = False
+
+    for curr_loc in exec_locations:
+        # only filter by thread if the trace has a thread column
+        if has_thread:
+            curr_process, curr_thread = curr_loc
+            filtered_df = enter_leave_df.loc[
+                (enter_leave_df["Process"] == curr_process)
+                & (enter_leave_df["Thread"] == curr_thread)
+            ]
+        else:
+            filtered_df = enter_leave_df.loc[(enter_leave_df["Process"] == curr_loc)]
+
+        curr_depth, callpath = 0, ""
+
+        """
+        Iterating over lists instead of
+        DataFrame columns is more efficient
+        """
+        df_indices = filtered_df.index.to_list()
+        function_names = filtered_df["Name"].to_list()
+        event_types = filtered_df["Event Type"].to_list()
+
+        # stacks used to iterate through the trace and add nodes to the cct
+        functions_stack, nodes_stack = [], []
+
+        # iterating over the events of the current thread's trace
+        for i in range(len(filtered_df)):
+            curr_df_index, evt_type, function_name = (
+                df_indices[i],
+                event_types[i],
+                function_names[i],
+            )
+
+            # encounter a new function through its entry point.
+            if evt_type == "Enter":
+                # add the function to the stack and get the call path
+                functions_stack.append(function_name)
+                callpath = "->".join(functions_stack)
+
+                # get the parent node of the function if it exists
+                parent_node = None if curr_depth == 0 else nodes_stack[-1]
+
+                if callpath in callpath_to_node:
+                    # don't create new node if callpath is in map
+                    curr_node = callpath_to_node[callpath]
+                else:
+                    # create new node if callpath isn't in map
+                    curr_node = Node(node_id, parent_node, curr_depth)
+                    callpath_to_node[callpath] = curr_node
+                    node_id += 1
+
+                    # add node as root or child of its
+                    # parent depending on current depth
+                    graph.add_root(
+                        curr_node
+                    ) if curr_depth == 0 else parent_node.add_child(curr_node)
+
+                # Update nodes stack, column, and current depth
+                nodes_stack.append(curr_node)
+                graph_nodes[curr_df_index] = curr_node
+                curr_depth += 1
+            else:
+                """
+                Pop node from top of stack once you
+                encounter the Leave event for a function
+                """
+                nodes_stack.pop()
+
+                # Update functions stack and current depth
+                functions_stack.pop()
+                curr_depth -= 1
+
+    # Update the Trace with the generated cct
+    events["Graph_Node"] = graph_nodes
+
+    return graph