Skip to content

deallocate otf2 bug-fix and set up test environment for otf2 with mul… #117

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
rocm-devops wants to merge 1 commit into
base: amd-staging
Choose a base branch
from
Open

deallocate otf2 bug-fix and set up test environment for otf2 with mul… #117

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
2 changes: 1 addition & 1 deletion source/lib/python/rocpd/libpyrocpd.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -676,7 +676,7 @@ PYBIND11_MODULE(libpyrocpd, pyrocpd)
auto sqlgen_otf2 = common::simple_timer{
fmt::format("OTF2 generation from {} SQL database(s)", data.size())};

uint16_t _process_counter = 0;
uint16_t _process_counter = 1;
for(auto obj : {data.connection})
{
auto* conn = rocpd::interop::get_connection(std::move(obj));
Expand Down
128 changes: 71 additions & 57 deletions source/lib/python/rocpd/source/otf2.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -139,6 +139,9 @@ archive_t* archive = nullptr;
auto flush_callbacks = OTF2_FlushCallbacks{pre_flush, post_flush};
OTF2_GlobalDefWriter* global_def_writer = nullptr; // shared between data bases (processes)

hash_map_t hash_data = {}; // shared definition dictionary between data bases and nodes
auto existing_hash = std::unordered_set<size_t>{};

enum rocprofiler_location_type_t
{
ROCPROFILER_AGENT_NO_TYPE = 0,
Expand Down Expand Up @@ -299,6 +302,16 @@ add_event(std::string_view name,
ROCP_FATAL << "otf2::add_event phase is not enter or exit";
}

void
add_write_string(size_t _hash, std::string_view _name_strv)
{
if(_hash > 0 && existing_hash.count(_hash) == 0)
{
OTF2_CHECK(OTF2_GlobalDefWriter_WriteString(global_def_writer, _hash, _name_strv.data()));
existing_hash.emplace(_hash);
}
};

void
setup(const rocprofiler::tool::output_config& cfg, uint64_t min_start, uint64_t max_fini)
{
Expand Down Expand Up @@ -345,28 +358,14 @@ setup(const rocprofiler::tool::output_config& cfg, uint64_t min_start, uint64_t

OTF2_CHECK(OTF2_GlobalDefWriter_WriteString(global_def_writer, 0, ""));

auto add_write_string = [](size_t _hash, std::string_view _name_strv) {
static auto _existing = std::unordered_set<size_t>{};
if(_hash > 0 && _existing.count(_hash) == 0)
{
OTF2_CHECK(
OTF2_GlobalDefWriter_WriteString(global_def_writer, _hash, _name_strv.data()));
_existing.emplace(_hash);
}
};

auto add_write_string_val = [&add_write_string](std::string_view _name_v) {
auto _hash_v = get_hash_id(_name_v);
add_write_string(_hash_v, _name_v);
return _hash_v;
};

//(must be shared between processes)
auto _attr_name = std::string_view{"category"};
auto _attr_desc = std::string_view{"tracing category"};
auto _attr_name = std::string_view{"category"};
auto _attr_name_hash = get_hash_id(_attr_name);
add_write_string(_attr_name_hash, _attr_name);

auto _attr_name_hash = add_write_string_val(_attr_name);
auto _attr_desc_hash = add_write_string_val(_attr_desc);
auto _attr_desc = std::string_view{"tracing category"};
auto _attr_desc_hash = get_hash_id(_attr_desc);
add_write_string(_attr_desc_hash, _attr_desc);

OTF2_CHECK(OTF2_GlobalDefWriter_WriteAttribute(
global_def_writer, 0, _attr_name_hash, _attr_desc_hash, OTF2_TYPE_STRING));
Expand Down Expand Up @@ -428,17 +427,16 @@ write_otf2(const OTF2Session& otf2_session,
const tool::generator<types::memory_copies>& memory_copy_gen,
const tool::generator<types::memory_allocation>& memory_allocation_gen)
{
const uint64_t _no_agent_handle = 0;
// std::numeric_limits<uint64_t>::max() - 1;
const auto& ocfg = otf2_session.config;

auto _app_ts = rocprofiler::tool::timestamps_t{process.start, process.fini};
const auto& ocfg = otf2_session.config;
auto _app_ts = rocprofiler::tool::timestamps_t{process.start, process.fini};

auto thread_event_info = std::map<pid_t, event_info>{};
auto agent_memcpy_info =
std::map<pid_t, std::map<uint64_t, event_info>>{}; // tid -> agent_handle ->evt
auto agent_memalloc_info =
std::map<pid_t, std::map<uint64_t, event_info>>{}; // // tid -> agent_handle ->evt
auto mem_dealloc_info = std::map<pid_t, event_info>{}; // // tid -> evt
auto mem_unknown_info = std::map<pid_t, event_info>{}; // // tid -> evt
auto agent_dispatch_info =
std::map<pid_t,
std::map<uint64_t, std::map<uint64_t, event_info>>>{}; // tid -> agent_handle
Expand Down Expand Up @@ -469,8 +467,6 @@ write_otf2(const OTF2Session& otf2_session,
thread_event_info.emplace(itr.tid, _evt_info);
}

auto _hash_data = hash_map_t{};

struct evt_data
{
rocprofiler_callback_phase_t phase = ROCPROFILER_CALLBACK_PHASE_NONE;
Expand Down Expand Up @@ -505,8 +501,8 @@ write_otf2(const OTF2Session& otf2_session,
for(const auto& itr : api_gen.get(ditr))
{
std::string _name = itr.name;
_hash_data.emplace(get_hash_id(_name),
region_info{_name, OTF2_REGION_ROLE_FUNCTION, OTF2_PARADIGM_HIP});
hash_data.emplace(get_hash_id(_name),
region_info{_name, OTF2_REGION_ROLE_FUNCTION, OTF2_PARADIGM_HIP});

auto& _evt_info = thread_event_info.at(itr.tid);
_evt_info.event_count += 1;
Expand Down Expand Up @@ -534,7 +530,7 @@ write_otf2(const OTF2Session& otf2_session,
for(const auto& itr : memory_copy_gen.get(ditr))
{
std::string _name = itr.name;
_hash_data.emplace(
hash_data.emplace(
get_hash_id(_name),
region_info{_name, OTF2_REGION_ROLE_DATA_TRANSFER, OTF2_PARADIGM_HIP});

Expand Down Expand Up @@ -583,7 +579,7 @@ write_otf2(const OTF2Session& otf2_session,

if(itr.type == "ALLOC")
{
_hash_data.emplace(
hash_data.emplace(
get_hash_id(_alloc_operation),
region_info{_alloc_operation, OTF2_REGION_ROLE_ALLOCATE, OTF2_PARADIGM_HIP});

Expand Down Expand Up @@ -615,18 +611,15 @@ write_otf2(const OTF2Session& otf2_session,
}
else if(itr.type == "FREE") //
{
_hash_data.emplace(
hash_data.emplace(
get_hash_id(_alloc_operation),
region_info{_alloc_operation, OTF2_REGION_ROLE_DEALLOCATE, OTF2_PARADIGM_HIP});
auto _evt_info = event_info{
location_base{process.pid, itr.tid, ROCPROFILER_AGENT_MEMORY_DEALLOC_TYPE}};

auto _evt_info = event_info{location_base{
process.pid, itr.tid, _no_agent_handle, ROCPROFILER_AGENT_MEMORY_DEALLOC_TYPE}};
_evt_info.name = fmt::format("Thread {}, Memory Deallocate (Free)", itr.tid);

agent_memalloc_info[itr.tid].emplace(_no_agent_handle, _evt_info);

_evt_info.event_count += 1;

mem_dealloc_info.emplace(itr.tid, _evt_info);
_data.emplace_back(evt_data{ROCPROFILER_CALLBACK_PHASE_ENTER,
_alloc_operation,
_evt_info.get_location(),
Expand All @@ -643,7 +636,7 @@ write_otf2(const OTF2Session& otf2_session,
auto _evt_info = event_info{location_base{process.pid, itr.tid}};
_evt_info.name = fmt::format("Thread {}, Memory Operation UNK", itr.tid);
_evt_info.event_count += 1;
agent_memalloc_info[itr.tid].emplace(_no_agent_handle, _evt_info);
mem_unknown_info.emplace(itr.tid, _evt_info);
_data.emplace_back(evt_data{ROCPROFILER_CALLBACK_PHASE_ENTER,
_alloc_operation,
_evt_info.get_location(),
Expand All @@ -663,8 +656,8 @@ write_otf2(const OTF2Session& otf2_session,
{
auto _name = fmt::format(
"{}", (ocfg.kernel_rename && !itr.region.empty()) ? itr.region : itr.name);
_hash_data.emplace(get_hash_id(_name),
region_info{_name, OTF2_REGION_ROLE_FUNCTION, OTF2_PARADIGM_HIP});
hash_data.emplace(get_hash_id(_name),
region_info{_name, OTF2_REGION_ROLE_FUNCTION, OTF2_PARADIGM_HIP});

const auto* _perfetto_name = rocprofiler::sdk::perfetto_category<
rocprofiler::sdk::category::kernel_dispatch>::name;
Expand Down Expand Up @@ -738,16 +731,13 @@ write_otf2(const OTF2Session& otf2_session,
}
OTF2_CHECK(OTF2_Archive_CloseDefFiles(archive));

for(const auto& itr : _hash_data)
for(const auto& itr : hash_data)
{
if(itr.first != 0)
if(itr.first != 0 && existing_hash.count(itr.first) == 0)
{
OTF2_CHECK(OTF2_GlobalDefWriter_WriteString(
global_def_writer, itr.first, itr.second.name.c_str()));
}

for(const auto& itr : _hash_data)
{
if(itr.first != 0)
OTF2_CHECK(OTF2_GlobalDefWriter_WriteRegion(global_def_writer,
itr.first,
itr.first,
Expand All @@ -759,16 +749,11 @@ write_otf2(const OTF2Session& otf2_session,
0,
0,
0));
}

auto add_write_string = [](size_t _hash, std::string_view _name) {
static auto _existing = std::unordered_set<size_t>{};
if(_hash > 0 && _existing.count(_hash) == 0)
{
OTF2_CHECK(OTF2_GlobalDefWriter_WriteString(global_def_writer, _hash, _name.data()));
_existing.emplace(_hash);
// Add to the list of processed definitions
existing_hash.emplace(itr.first);
}
};
}

for(const auto& itr : _attr_str)
add_write_string(itr.first, itr.second);
Expand Down Expand Up @@ -873,11 +858,40 @@ write_otf2(const OTF2Session& otf2_session,
_hash,
OTF2_LOCATION_TYPE_ACCELERATOR_STREAM,
2 * evt.event_count, // # events
agent_handle // location group
));
agent_handle) // location group
);
}
}

// Mem-free events
for(auto& [tid, evt] : mem_dealloc_info)
{
auto _hash = get_hash_id(evt.name);

add_write_string(_hash, evt.name);
OTF2_CHECK(OTF2_GlobalDefWriter_WriteLocation(global_def_writer,
evt.id(), // id
_hash,
OTF2_LOCATION_TYPE_UNKNOWN,
2 * evt.event_count,
tree_node_id // location group
));
}

// Mem-unknown events
for(auto& [tid, evt] : mem_unknown_info)
{
auto _hash = get_hash_id(evt.name);

add_write_string(_hash, evt.name);
OTF2_CHECK(OTF2_GlobalDefWriter_WriteLocation(global_def_writer,
evt.id(), // id
_hash,
OTF2_LOCATION_TYPE_UNKNOWN,
2 * evt.event_count,
tree_node_id));
}

// Dispatch Events
for(auto& [tid, itr] : agent_dispatch_info)
{
Expand Down
46 changes: 24 additions & 22 deletions tests/pytest-packages/pytest_utils/otf2_reader.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,11 +72,13 @@ def __str__(self):


class OTF2Reader:
"""Read in perfetto protobuf output"""
"""Read in otf2 output"""

def __init__(self, filename):
self.filename = filename if isinstance(filename, (list, tuple)) else [filename]

# returns the the map *reader -> [data_frame], where *reader is created per input otf2 file,
# and each data_frame in [data_frame] corresponds to a rocpd data base (indexed by tree_node) which contributed to the otf2 file
def read(self):
def _read_trace(trace_name):
trace = otf2.reader.Reader(trace_name)
Expand Down Expand Up @@ -150,20 +152,21 @@ def _read_trace(trace_name):
f"Modified length ({_mlen}) != Expected length({_elen}) for {event} at {location}"
)

data = {
"system_tree_node": [],
"location_group": [],
"location": [],
"region": [],
"attributes": [],
"depth": [],
"name": [],
"category": [],
"start_ts": [],
"end_ts": [],
}

process_dfs = []
for tree, lgitr in call_stack.items():
data = {
"system_tree_node": [],
"location_group": [],
"location": [],
"region": [],
"attributes": [],
"depth": [],
"name": [],
"category": [],
"start_ts": [],
"end_ts": [],
}

for group, gitr in lgitr.items():
for loc, ritr in gitr.items():
for region in ritr:
Expand All @@ -177,21 +180,20 @@ def _read_trace(trace_name):
data["name"] += [region.name]
data["start_ts"] += [region.enter_nsec]
data["end_ts"] += [region.leave_nsec]
process_dfs += [pd.DataFrame.from_dict(data)]

return (trace, pd.DataFrame.from_dict(data))
return (trace, process_dfs)

readers = []
df = pd.DataFrame()
ret_val = {}
for itr in self.filename:
_reader, _df = _read_trace(itr)
readers += [_reader]
df = pd.concat([df, _df])
_reader, _process_dfs = _read_trace(itr)
ret_val[_reader] = _process_dfs

return (df, readers)
return ret_val


def read_trace(filename):
data = OTF2Reader(filename).read()[0]
data = list(OTF2Reader(filename).read().values())[0][0]

print(f"\nDATA:\n{data}")

Expand Down
Loading