Skip to content

Don't aggregate kernel records into hipLaunch parent #182

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
michaelmckinsey1 wants to merge 1 commit into
base: develop
Choose a base branch
from
+220 −89
Open

Don't aggregate kernel records into hipLaunch parent #182

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
234 changes: 145 additions & 89 deletions hatchet/readers/caliper_native_reader.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,8 @@ def __raise_cali_type_error(msg):
class CaliperNativeReader:
"""Read in a native `.cali` file using Caliper's python reader."""

_kernel_placeholder_labels = {"|-"}

__cali_type_dict = {
"inv": lambda dummy: __raise_cali_type_error(
"Caliper type 'inv' is unsupported in Hatchet"
Expand Down Expand Up @@ -84,6 +86,135 @@ def __init__(
if isinstance(self.string_attributes, str):
self.string_attributes = [self.string_attributes]

def _find_unique_child_callpath(self, parent_callpath, node_type=None):
"""Return a unique child callpath for a known parent, if one exists."""
matches = []

for callpath, hnode in self.callpath_to_node.items():
if len(callpath) != len(parent_callpath) + 1:
continue
if callpath[:-1] != parent_callpath:
continue
if node_type is not None and hnode.frame["type"] != node_type:
continue
matches.append(callpath)

if len(matches) == 1:
return matches[0]

return None

def _resolve_record_node_info(self, record, ctx="path"):
"""Resolve a Caliper record to the node it should update/create."""
if ctx not in record:
return None

if not isinstance(record[ctx], list):
node_label = record[ctx]
return {
"label": node_label,
"callpath": tuple([node_label]),
"parent_callpath": tuple(),
"type": "function",
}

record_path = list(record[ctx])

# specify how to parse cupti records
if "cupti.activity.kind" in record:
if record["cupti.activity.kind"] == "kernel":
node_label = record["cupti.kernel.name"]
node_type = "kernel"
elif record["cupti.activity.kind"] == "memcpy":
node_label = record["cupti.activity.kind"]
node_type = "memcpy"
else:
node_label = record["cupti.activity.kind"]
node_type = "function"

node_callpath = tuple(record_path + [node_label])
return {
"label": node_label,
"callpath": node_callpath,
"parent_callpath": node_callpath[:-1],
"type": node_type,
}

# rocm records
if "rocm.activity" in record:
if record["rocm.activity"] == "KERNEL_DISPATCH_COMPLETE":
node_label = record["rocm.kernel.name"]
node_type = "kernel"
node_callpath = tuple(record_path + [node_label])
elif "rocm.api" in record and "hipMemcpy" in record["rocm.api"]:
node_label = record["rocm.activity"]
node_type = "memcpy"
# There is an extra placeholder path node for these records.
node_callpath = tuple(record_path[:-1] + [node_label])
else:
node_label = record["rocm.activity"]
node_type = "function"
node_callpath = tuple(record_path + [node_label])

return {
"label": node_label,
"callpath": node_callpath,
"parent_callpath": node_callpath[:-1],
"type": node_type,
}

# Some ROCm summary rows only carry the kernel name and counters.
if "rocm.kernel.name" in record and record["rocm.kernel.name"]:
node_label = record["rocm.kernel.name"]
node_callpath = tuple(record_path + [node_label])
return {
"label": node_label,
"callpath": node_callpath,
"parent_callpath": node_callpath[:-1],
"type": "kernel",
}

# Other summary rows only carry the placeholder path. Reuse the unique
# kernel child created from the dispatch record when possible.
if record_path and record_path[-1] in self._kernel_placeholder_labels:
existing_kernel_callpath = self._find_unique_child_callpath(
tuple(record_path), node_type="kernel"
)
if existing_kernel_callpath is not None:
return {
"label": existing_kernel_callpath[-1],
"callpath": existing_kernel_callpath,
"parent_callpath": existing_kernel_callpath[:-1],
"type": "kernel",
}

# Sampling
if "module#cali.sampler.pc" in record:
node_label = record["source.function#cali.sampler.pc"]
node_callpath = tuple(record_path)[:-1] + (node_label,)
return {
"label": node_label,
"callpath": node_callpath,
"parent_callpath": node_callpath[:-1],
"type": "function",
}

if len(record_path) > 0:
node_label = record_path[-1]
node_callpath = tuple(record_path)
parent_callpath = node_callpath[:-1]
else:
node_label = "unknown_context"
node_callpath = tuple("unknown_context")
parent_callpath = tuple("")

return {
"label": node_label,
"callpath": node_callpath,
"parent_callpath": parent_callpath,
"type": "function",
}

def _create_metric_df(self, metrics, sampling=False):
"""Make a list of metric columns and create a dataframe, group by node"""
for col in self.record_data_cols:
Expand All @@ -106,7 +237,7 @@ def _create_metric_df(self, metrics, sampling=False):
elif np.issubdtype(
df_metrics[column].dtype, np.number
): # Numeric columns
aggregation_functions[column] = lambda x: x.sum(skipna=False)
aggregation_functions[column] = lambda x: x.sum(min_count=1)
else: # Non-numeric columns
aggregation_functions[column] = lambda x: tuple(set(x))

Expand Down Expand Up @@ -153,52 +284,22 @@ def read_metrics(self, ctx="path"):
cur_timestep = next_timestep

node_dict = {}
if ctx in record:
node_info = self._resolve_record_node_info(record, ctx)
if node_info is not None:
# only parse records that have spot.channel=regionprofile or no
# spot.channel attribute
if (
"spot.channel" in record
and record["spot.channel"] == "regionprofile"
) or "spot.channel" not in record:
# get the node label and callpath for the record
if isinstance(record[ctx], list):
# specify how to parse cupti records
if "cupti.activity.kind" in record:
if record["cupti.activity.kind"] == "kernel":
node_label = record["cupti.kernel.name"]
node_callpath = tuple(record[ctx] + [node_label])
elif record["cupti.activity.kind"] == "memcpy":
node_label = record["cupti.activity.kind"]
node_callpath = tuple(record[ctx] + [node_label])
# rocm records
elif "rocm.activity" in record:
if record["rocm.activity"] == "KERNEL_DISPATCH_COMPLETE":
node_label = record["rocm.kernel.name"]
node_callpath = tuple(record[ctx] + [node_label])
else:
node_label = record["rocm.activity"]
node_callpath = tuple(record[ctx] + [node_label])
# Sampling
elif "module#cali.sampler.pc" in record:
node_label = record["source.function#cali.sampler.pc"]
node_callpath = tuple(record[ctx])[:-1] + (node_label,)
sampling = True
else:
if len(record[ctx]) > 0:
node_label = record[ctx][-1]
node_callpath = tuple(record[ctx])
else:
node_label = "unknown_context"
node_callpath = tuple("unknown_context")
else:
node_label = record[ctx]
node_callpath = tuple([record[ctx]])
if "module#cali.sampler.pc" in record:
sampling = True

if "spot.channel" in record:
node_dict["spot.channel"] = record["spot.channel"]

# get node nid based on callpath
node_dict["nid"] = self.callpath_to_idx.get(node_callpath)
node_dict["nid"] = self.callpath_to_idx.get(node_info["callpath"])

for item in record.keys():
if item != ctx:
Expand Down Expand Up @@ -299,62 +400,17 @@ def _create_parent(child_node, parent_callpath):
order = -1

for record in records:
node_label = ""
if ctx in record:
node_info = self._resolve_record_node_info(record, ctx)
if node_info is not None:
if (
"spot.channel" in record
and record["spot.channel"] == "regionprofile"
) or "spot.channel" not in record:
# if it's a list, then it's a callpath
if isinstance(record[ctx], list):
# specify how to parse cupti records
if "cupti.activity.kind" in record:
if record["cupti.activity.kind"] == "kernel":
node_label = record["cupti.kernel.name"]
node_callpath = tuple(record[ctx] + [node_label])
parent_callpath = node_callpath[:-1]
node_type = "kernel"
elif record["cupti.activity.kind"] == "memcpy":
node_label = record["cupti.activity.kind"]
node_callpath = tuple(record[ctx] + [node_label])
parent_callpath = node_callpath[:-1]
node_type = "memcpy"
else:
Exception("Haven't seen this activity kind yet")
# rocm records
elif "rocm.activity" in record:
if record["rocm.activity"] == "KERNEL_DISPATCH_COMPLETE":
node_label = record["rocm.kernel.name"]
node_callpath = tuple(record[ctx] + [node_label])
parent_callpath = node_callpath[:-1]
node_type = "kernel"
elif "hipMemcpy" in record["rocm.api"]:
node_label = record["rocm.activity"]
# Theres going to be an extra record at the end that we must remove
if len(record[ctx]) > 0:
record[ctx].pop()
node_callpath = tuple(record[ctx] + [node_label])
parent_callpath = node_callpath[:-1]
node_type = "memcpy"
else:
Exception("Haven't seen this activity kind yet")
# Sampling
elif "module#cali.sampler.pc" in record:
node_label = record["source.function#cali.sampler.pc"]
node_callpath = tuple(record[ctx])[:-1] + (node_label,)
parent_callpath = node_callpath[:-1]
node_type = "function"
else:
if len(record[ctx]) > 0:
node_label = record[ctx][-1]
node_callpath = tuple(record[ctx])
parent_callpath = node_callpath[:-1]
node_type = "function"
else:
node_label = "unknown_context"
node_callpath = tuple("unknown_context")
parent_callpath = tuple("")
node_type = "function"
node_label = node_info["label"]
node_callpath = node_info["callpath"]
parent_callpath = node_info["parent_callpath"]
node_type = node_info["type"]

hnode = self.callpath_to_node.get(node_callpath)

Expand Down Expand Up @@ -403,8 +459,8 @@ def _create_parent(child_node, parent_callpath):

# if it's a string, then it's a root
else:
root_label = record[ctx]
root_callpath = tuple([root_label])
root_label = node_info["label"]
root_callpath = node_info["callpath"]

if root_callpath not in self.callpath_to_node:
# create the root since it doesn't exist
Expand Down
75 changes: 75 additions & 0 deletions hatchet/tests/caliper.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@

from hatchet import GraphFrame
from hatchet.readers.caliper_reader import CaliperReader
from hatchet.readers.caliper_native_reader import CaliperNativeReader
from hatchet.util.executable import which

caliperreader_avail = True
Expand Down Expand Up @@ -886,6 +887,80 @@ def test_sw4_cuda_summary_from_caliperreader(
assert type(gf.metadata["cali.channel"]) == str


class _FakeCaliperAttribute:
def __init__(self, attr_type, is_value=False, alias=None):
self._attr_type = attr_type
self._is_value = is_value
self._alias = alias

def attribute_type(self):
return self._attr_type

def is_value(self):
return self._is_value

def get(self, key):
if key == "attribute.alias":
return self._alias
return None


class _FakeCaliperReader:
def __init__(self, records, attributes):
self.records = records
self.globals = {}
self._attributes = attributes

def attribute(self, name):
return self._attributes[name]


def test_rocm_kernel_records_merge_on_single_kernel_node():
kernel_name = "void RAJA::launch..."
fake_reader = _FakeCaliperReader(
records=[
{
"path": ["hipLaunchKernel", "|-"],
"rocm.activity": "KERNEL_DISPATCH_COMPLETE",
"rocm.kernel.name": kernel_name,
"time.duration": 5.0,
},
{
"path": ["hipLaunchKernel", "|-"],
"rocm.kernel.name": kernel_name,
"SQ_INSTS": 42.0,
},
{
"path": ["hipLaunchKernel", "|-"],
"MaxPerfectLoopDimensions": 2,
},
],
attributes={
"path": _FakeCaliperAttribute("string"),
"rocm.activity": _FakeCaliperAttribute("string"),
"rocm.kernel.name": _FakeCaliperAttribute("string"),
"time.duration": _FakeCaliperAttribute("double", is_value=True),
"SQ_INSTS": _FakeCaliperAttribute("double", is_value=True),
"MaxPerfectLoopDimensions": _FakeCaliperAttribute(
"int", is_value=True
),
},
)

gf = CaliperNativeReader(
fake_reader, native=True, string_attributes=[], node_ordering=False
).read()

df = gf.dataframe.reset_index()

kernel_rows = df[df["name"] == kernel_name]
assert len(kernel_rows) == 1
assert kernel_rows.iloc[0]["SQ_INSTS"] == 42.0
assert kernel_rows.iloc[0]["MaxPerfectLoopDimensions"] == 2
assert kernel_rows.iloc[0]["time.duration"] == 5.0
assert "|-" not in df["name"].tolist()


def test_graphframe_timeseries_lulesh_from_file(caliper_timeseries_cali):
"""Sanity check the timeseries Caliper reader by examining a known input."""

Expand Down
Loading