AI/rocm-systems
2
0
Rozštěpit 0
Zdrojový kód Úkoly Pull requesty Akce Balíčky Projekty Vydání Wiki Aktivita

Change how cache manager handles child process trace cache for rocpd (#1033)

Procházet zdrojové kódy 
* Change how cache manager handles child process trace cache

* Sampling and backtrace metrics to cache

* Apply cmake formatting

* Fix parsing of metadata json

* Code clean up

* Fix build nlohmann json from source

* Fix storage parsed finished callback

* Revert sampling for child process

* Change cache file name generating

* Fix thread start stop

* Fix process start end timestamp

* Applied suggestions from code review

* Try with late start of flushing task thread

* Change dockerfiles for ci

* Revert changes on github workflows

* Remove json_fwd.hpp include

* fix dump

* Build nlohmann/json by default

Signed-off-by: David Galiffi <David.Galiffi@amd.com>

* Update location of build artifacts for nlohmann/json

Signed-off-by: David Galiffi <David.Galiffi@amd.com>

* Revert use_output_suffix

* Remove unused logs

* Fix cache store inside counter due to structure change

* Remove decode tests from debian ci

* Fix issue where all databases have the same UUID (#1499)

Co-authored-by: Aleksandar Djordjevic <adjordje@amd.com>

* Removing the cpack and install steps to save space

* Revert "Remove decode tests from debian ci"

This reverts commit ddabf6dd142dcf438e6b8997b8abe86f2c868468.

* Revert "Removing the cpack and install steps to save space"

This reverts commit 973da3a1ba99d99d529af5269d30e177092f9bfa.

* Add prepare-runner job as dependency to clean up the space

* Fix formatting

* Free up even more space

* Remove verbose for workflows

* remove hw_counters from ext_data

* move space clean up inside container

* try to remove external folder to free up space

* Check space

* Refactor Cleanup to it's own step

---------

Signed-off-by: David Galiffi <David.Galiffi@amd.com>
Co-authored-by: David Galiffi <David.Galiffi@amd.com>
Co-authored-by: Aleksandar Djordjevic <aleksandar.djordjevic@amd.com>
Co-authored-by: Aleksandar Djordjevic <adjordje@amd.com>
Tento commit je obsažen v:
Milan Radosavljevic
2025-10-24 17:47:15 +02:00
odevzdal GitHub
rodič 4f075902fc
revize 8806be162c
46 změnil soubory, kde provedl 1502 přidání a 1014 odebrání
Stáhněte soubor záplaty Stáhněte rozdílový soubor Rozbalit všechny soubory Sbalit všechny soubory
projects/rocprofiler-systems/source/lib/rocprof-sys/library/sampling.cpp
+133 -258
Zobrazit soubor
@@ -29,7 +29,6 @@
#include "core/node_info.hpp"
#include "core/perf.hpp"
#include "core/rocpd/data_processor.hpp"
#include "core/rocpd/json.hpp"
#include "core/state.hpp"
#include "core/trace_cache/cache_manager.hpp"
#include "core/utility.hpp"
@@ -73,6 +72,8 @@
#include <timemory/utility/types.hpp>
#include <timemory/variadic.hpp>
#include <nlohmann/json.hpp>
#include <array>
#include <chrono>
#include <condition_variable>
@@ -168,21 +169,21 @@ get_category_track_name(uint64_t tid)
std::string
generate_call_stack_json(const tim::unwind::processed_entry& stack_entry)
{
auto call_stack = ::rocpd::json::create();
nlohmann::json call_stack;
call_stack->set("name", std::string(demangle(stack_entry.name)));
call_stack->set("pc", as_hex(stack_entry.address));
call_stack->set("file", std::string(stack_entry.location));
call_stack["name"] = std::string(demangle(stack_entry.name));
call_stack["pc"] = as_hex(stack_entry.address);
call_stack["file"] = std::string(stack_entry.location);
return call_stack->to_string();
return call_stack.dump();
}
std::string
generate_line_info_json(const tim::unwind::processed_entry& line_info_entry)
{
auto line_info = ::rocpd::json::create();
line_info->set("line_address", as_hex(line_info_entry.line_address));
line_info->set("name", std::string(demangle(line_info_entry.name)));
nlohmann::json line_info;
line_info["line_address"] = as_hex(line_info_entry.line_address);
line_info["name"] = std::string(demangle(line_info_entry.name));
if(line_info_entry.lineinfo && !line_info_entry.lineinfo.lines.empty())
{
@@ -190,43 +191,15 @@ generate_line_info_json(const tim::unwind::processed_entry& line_info_entry)
std::reverse(_lines.begin(), _lines.end());
for(const auto& line : _lines)
{
auto inlined = ::rocpd::json::create();
inlined->set("name", std::string(demangle(line.name)));
inlined->set("location", std::string(line.location));
inlined->set("line", std::to_string(line.line));
line_info->set("inlined", inlined);
nlohmann::json inlined;
inlined["name"] = std::string(demangle(line.name));
inlined["location"] = std::string(line.location);
inlined["line"] = std::to_string(line.line);
line_info["inlined"] = inlined;
}
}
return line_info->to_string();
}
std::string
generate_hw_counter_json(int64_t _tid, const backtrace_metrics& metrics)
{
auto extdata = ::rocpd::json::create();
if(!metrics.get_hw_counters().empty())
{
auto _labels = backtrace_metrics::get_hw_counter_labels(_tid);
auto _hw_cnt_vals = metrics.get_hw_counters();
auto hw_counters = ::rocpd::json::create();
for(size_t i = 0; i < _labels.size(); ++i)
{
hw_counters->set(_labels.at(i), _hw_cnt_vals.at(i));
}
extdata->set("hw_counters", hw_counters);
}
return extdata->to_string();
}
rocpd::data_processor&
get_data_processor()
{
return rocpd::data_processor::get_instance();
return line_info.dump();
}
template <typename Category>
@@ -237,11 +210,13 @@ get_track_name(const thread_info& _thread_info)
size_t sequent_value = _thread_info.index_data->sequent_value;
constexpr auto sample_type =
std::is_same_v<Category, category::timer_sampling> ? "Timer" : "Overflow";
return JOIN(" ", "Thread", sequent_value, sample_type, "(S)", thread_id);
std::stringstream name_ss;
name_ss << "Thread " << sequent_value << " " << sample_type << " (S) " << thread_id;
return name_ss.str();
}
void
rocpd_initialize_sampling_category()
metadata_initialize_sampling_category()
{
static bool _is_initialized = false;
if(_is_initialized) return;
@@ -257,7 +232,7 @@ rocpd_initialize_sampling_category()
}
void
rocpd_initialize_thread_info(size_t tid)
metadata_initialize_thread_info(size_t tid)
{
const auto& _thread_info = thread_info::get(tid, SequentTID);
ROCPROFSYS_CI_THROW(!_thread_info, "No valid thread info for tid=%li\n", tid);
@@ -271,7 +246,7 @@ rocpd_initialize_thread_info(size_t tid)
}
void
rocpd_init_track(int64_t tid)
metadata_initialize_track(int64_t tid)
{
const auto& _thread_info = thread_info::get(tid, SequentTID);
ROCPROFSYS_CI_THROW(!_thread_info, "No valid thread info for tid=%li\n", tid);
@@ -290,22 +265,98 @@ rocpd_init_track(int64_t tid)
{ _overflow_track_name, thread_id, "{}" });
}
template <typename Category>
void
rocpd_insert_region(size_t thread_id, size_t start_time, size_t end_time, size_t name_id,
const char* track, const char* call_stack = "{}",
const char* line_info = "{}", const char* extdata = "{}")
// Added
struct timer_sampling_data
{
auto& data_processor = get_data_processor();
auto& n_info = node_info::get_instance();
auto string_primary_key = data_processor.insert_string(trait::name<Category>::value);
int64_t m_tid = -1;
uint64_t m_beg = 0;
uint64_t m_end = 0;
std::vector<tim::unwind::processed_entry> m_stack = {};
backtrace_metrics m_metrics = {};
};
auto event_id = data_processor.insert_event(string_primary_key, 0, 0, 0, call_stack,
line_info, extdata);
struct overflow_sampling_data
{
int64_t m_tid = -1;
uint64_t m_beg = 0;
uint64_t m_end = 0;
std::vector<tim::unwind::processed_entry> m_stack = {};
};
data_processor.insert_region(n_info.id, getpid(), thread_id, start_time, end_time,
name_id, event_id);
data_processor.insert_sample(track, start_time, event_id);
std::vector<timer_sampling_data>
parse_timer_data(int64_t _tid, const bundle_t* _init,
const std::vector<bundle_t*>& _data);
std::vector<overflow_sampling_data>
parse_overflow_data(int64_t _tid, const bundle_t*, const std::vector<bundle_t*>& _data);
// TODO: should we remove _tid? it's inside timer_data and overflow_data
void
cache_sampling_data(int64_t _tid, const std::vector<timer_sampling_data>& _timer_data,
const std::vector<overflow_sampling_data>& _overflow_data)
{
ROCPROFSYS_VERBOSE(3 || get_debug_sampling(),
"[%li] Storing sampling data to trace cache...\n", _tid);
const auto& _thread_info = thread_info::get(_tid, SequentTID);
ROCPROFSYS_CI_THROW(!_thread_info, "No valid thread info for tid=%li\n", _tid);
if(!_thread_info) return;
// Store timer sampling data
for(const auto& itr : _timer_data)
{
if(!_thread_info->is_valid_lifetime({ itr.m_beg, itr.m_end })) continue;
for(const auto& iitr : itr.m_stack)
{
auto _name = std::string(demangle(iitr.name));
auto _track_name = get_track_name<category::timer_sampling>(*_thread_info);
auto _call_stack = generate_call_stack_json(iitr);
auto _line_info = generate_line_info_json(iitr);
trace_cache::get_buffer_storage().store(
trace_cache::entry_type::backtrace_region_sample,
static_cast<uint32_t>(ROCPROFSYS_CATEGORY_TIMER_SAMPLING),
static_cast<uint64_t>(_thread_info->index_data->system_value),
_track_name.c_str(), _name.c_str(), itr.m_beg, itr.m_end,
trait::name<category::timer_sampling>::value, _call_stack.c_str(),
_line_info.c_str(), "{}");
}
}
auto _overflow_event =
get_setting_value<std::string>("ROCPROFSYS_SAMPLING_OVERFLOW_EVENT").value_or("");
if(!_overflow_event.empty())
{
const auto _overflow_prefix = std::string_view{ "PERF_COUNT_" };
const auto _overflow_pos = _overflow_event.find(_overflow_prefix);
if(_overflow_pos != std::string::npos)
_overflow_event =
_overflow_event.substr(_overflow_pos + _overflow_prefix.length());
}
for(const auto& itr : _overflow_data)
{
if(!_thread_info->is_valid_lifetime({ itr.m_beg, itr.m_end })) continue;
for(const auto& iitr : itr.m_stack)
{
auto _name = std::string(demangle(iitr.name));
auto _track_name = get_track_name<category::overflow_sampling>(*_thread_info);
auto _call_stack = generate_call_stack_json(iitr);
auto _line_info = generate_line_info_json(iitr);
trace_cache::get_buffer_storage().store(
trace_cache::entry_type::backtrace_region_sample,
static_cast<uint32_t>(ROCPROFSYS_CATEGORY_OVERFLOW_SAMPLING),
static_cast<uint64_t>(_thread_info->index_data->system_value),
_track_name.c_str(), _name.c_str(), itr.m_beg, itr.m_end,
trait::name<category::overflow_sampling>::value, _call_stack.c_str(),
_line_info.c_str(), "{}");
}
}
}
auto&
@@ -871,9 +922,9 @@ configure(bool _setup, int64_t _tid)
}
}
}
rocpd_initialize_sampling_category();
rocpd_initialize_thread_info(_tid);
rocpd_init_track(_tid);
metadata_initialize_sampling_category();
metadata_initialize_thread_info(_tid);
metadata_initialize_track(_tid);
*_running = true;
sampling::get_sampler_init(_tid)->sample();
@@ -937,28 +988,11 @@ configure(bool _setup, int64_t _tid)
return (_signal_types) ? *_signal_types : std::set<int>{};
}
struct timer_sampling_data
{
int64_t m_tid = -1;
uint64_t m_beg = 0;
uint64_t m_end = 0;
std::vector<tim::unwind::processed_entry> m_stack = {};
backtrace_metrics m_metrics = {};
};
struct overflow_sampling_data
{
int64_t m_tid = -1;
uint64_t m_beg = 0;
uint64_t m_end = 0;
std::vector<tim::unwind::processed_entry> m_stack = {};
};
std::vector<timer_sampling_data>
post_process_timer_data(int64_t, const bundle_t*, const std::vector<bundle_t*>&);
parse_timer_data(int64_t, const bundle_t*, const std::vector<bundle_t*>&);
std::vector<overflow_sampling_data>
post_process_overflow_data(int64_t, const bundle_t*, const std::vector<bundle_t*>&);
parse_overflow_data(int64_t, const bundle_t*, const std::vector<bundle_t*>&);
void
post_process_perfetto(int64_t, const std::vector<timer_sampling_data>&,
@@ -969,8 +1003,9 @@ post_process_timemory(int64_t, const std::vector<timer_sampling_data>&,
const std::vector<overflow_sampling_data>&);
void
post_process_rocpd(int64_t _tid, const std::vector<timer_sampling_data>& _timer_data,
const std::vector<overflow_sampling_data>& _overflow_data);
store_sampling_data_in_cache(int64_t _tid,
const std::vector<timer_sampling_data>& _timer_data,
const std::vector<overflow_sampling_data>& _overflow_data);
auto static_strings = std::set<std::string>{};
@@ -1145,12 +1180,12 @@ post_process()
"Sampler data for thread %lu has %zu valid entries...\n",
i, _data.size());
auto _timer_data = post_process_timer_data(i, _init, _data);
auto _overflow_data = post_process_overflow_data(i, _init, _data);
auto _timer_data = parse_timer_data(i, _init, _data);
auto _overflow_data = parse_overflow_data(i, _init, _data);
if(get_use_perfetto()) post_process_perfetto(i, _timer_data, _overflow_data);
if(get_use_timemory()) post_process_timemory(i, _timer_data, _overflow_data);
if(get_use_rocpd()) post_process_rocpd(i, _timer_data, _overflow_data);
store_sampling_data_in_cache(i, _timer_data, _overflow_data);
}
else
{
@@ -1191,8 +1226,7 @@ post_process()
namespace
{
std::vector<timer_sampling_data>
post_process_timer_data(int64_t _tid, const bundle_t* _init,
const std::vector<bundle_t*>& _data)
parse_timer_data(int64_t _tid, const bundle_t* _init, const std::vector<bundle_t*>& _data)
{
auto _results = std::vector<timer_sampling_data>{};
@@ -1238,8 +1272,7 @@ post_process_timer_data(int64_t _tid, const bundle_t* _init,
}
std::vector<overflow_sampling_data>
post_process_overflow_data(int64_t _tid, const bundle_t*,
const std::vector<bundle_t*>& _data)
parse_overflow_data(int64_t _tid, const bundle_t*, const std::vector<bundle_t*>& _data)
{
auto _results = std::vector<overflow_sampling_data>{};
@@ -1704,66 +1737,7 @@ post_process_timemory(int64_t _tid, const std::vector<timer_sampling_data>& _tim
}
void
rocpd_post_process_overflow_data(
int64_t _tid, const std::vector<overflow_sampling_data>& _overflow_data)
{
auto& data_processor = get_data_processor();
const auto& _thread_info = thread_info::get(_tid, SequentTID);
ROCPROFSYS_CI_THROW(!_thread_info, "No valid thread info for tid=%li\n", _tid);
if(!_thread_info) return;
auto _overflow_event =
get_setting_value<std::string>("ROCPROFSYS_SAMPLING_OVERFLOW_EVENT").value_or("");
if(!_overflow_event.empty() && !_overflow_data.empty())
{
auto _beg_ns = std::max(_overflow_data.front().m_beg, _thread_info->get_start());
auto _end_ns = std::min(_overflow_data.back().m_end, _thread_info->get_stop());
const auto _overflow_prefix = std::string_view{ "PERF_COUNT_" };
const auto _overflow_pos = _overflow_event.find(_overflow_prefix);
if(_overflow_pos != std::string::npos)
_overflow_event =
_overflow_event.substr(_overflow_pos + _overflow_prefix.length());
const auto* _main_name =
static_strings.emplace(join(" ", _overflow_event, "samples [rocprof-sys]"))
.first->c_str();
auto main_name_id = data_processor.insert_string(_main_name);
size_t thread_id = _thread_info->index_data->system_value;
auto thread_primary_key = data_processor.map_thread_id_to_primary_key(thread_id);
const auto _track_name =
get_track_name<category::overflow_sampling>(*_thread_info);
rocpd_insert_region<category::overflow_sampling>(
thread_primary_key, _beg_ns, _end_ns, main_name_id, _track_name.c_str());
for(const auto& itr : _overflow_data)
{
auto _beg = itr.m_beg;
auto _end = itr.m_end;
if(!_thread_info->is_valid_lifetime({ _beg, _end })) continue;
for(const auto& iitr : itr.m_stack)
{
const auto* _name =
static_strings.emplace(demangle(iitr.name)).first->c_str();
auto name_id = data_processor.insert_string(_name);
rocpd_insert_region<category::overflow_sampling>(
thread_primary_key, _beg, _end, name_id, _track_name.c_str(),
generate_call_stack_json(iitr).c_str(),
generate_line_info_json(iitr).c_str());
}
}
}
}
void
rocpd_post_process_backtrace_metrics(
cache_backtrace_metrics(
[[maybe_unused]] int64_t _tid,
[[maybe_unused]] const std::vector<timer_sampling_data>& _timer_data)
{
@@ -1779,121 +1753,22 @@ rocpd_post_process_backtrace_metrics(
{
ROCPROFSYS_VERBOSE(3 || get_debug_sampling(),
"[%li] Post-processing metrics for rocpd...\n", _tid);
backtrace_metrics::init_rocpd(_tid, _valid_metrics); // move to setup
backtrace_metrics::init_cache(_tid, _valid_metrics); // move to setup
for(const auto& itr : _timer_data)
itr.m_metrics.post_process_rocpd(_tid, 0.5 * (itr.m_beg + itr.m_end));
backtrace_metrics::fini_rocpd(_tid, _valid_metrics);
itr.m_metrics.cache_backtrace_data(_tid, 0.5 * (itr.m_beg + itr.m_end));
}
#endif
}
void
rocpd_post_process_timer_data(
[[maybe_unused]] int64_t _tid,
[[maybe_unused]] const std::vector<timer_sampling_data>& _timer_data)
store_sampling_data_in_cache(
[[maybe_unused]] int64_t _tid,
[[maybe_unused]] const std::vector<timer_sampling_data>& _timer_data,
[[maybe_unused]] const std::vector<overflow_sampling_data>& _overflow_data)
{
#if ROCPROFSYS_USE_ROCM > 0
auto& data_processor = get_data_processor();
const auto& _thread_info = thread_info::get(_tid, SequentTID);
ROCPROFSYS_CI_THROW(!_thread_info, "No valid thread info for tid=%li\n", _tid);
if(!_thread_info) return;
if(!_timer_data.empty())
{
rocpd_post_process_backtrace_metrics(_tid, _timer_data);
auto _beg_ns = std::max(_timer_data.front().m_beg, _thread_info->get_start());
auto _end_ns = std::min(_timer_data.back().m_end, _thread_info->get_stop());
const auto _track_name = get_track_name<category::timer_sampling>(*_thread_info);
auto thread_primary_key = data_processor.map_thread_id_to_primary_key(
_thread_info->index_data->system_value);
const auto main_name_id = data_processor.insert_string("samples [rocprof-sys]");
rocpd_insert_region<category::timer_sampling>(
thread_primary_key, _beg_ns, _end_ns, main_name_id, _track_name.c_str());
auto _labels = backtrace_metrics::get_hw_counter_labels(_tid);
for(const auto& itr : _timer_data)
{
size_t _ncount = 0;
uint64_t _beg = itr.m_beg;
uint64_t _end = itr.m_end;
if(!_thread_info->is_valid_lifetime({ _beg, _end })) continue;
for(const auto& iitr : itr.m_stack)
{
auto _ncur = _ncount++;
// the begin/end + HW counters will be same for entire call-stack so only
// annotate the top and the bottom functions to keep the data consumption
// low
bool _include_common = (_ncur == 0 || _ncur + 1 == itr.m_stack.size());
// Only annotate HW counters when first or last and HW counters are not
// empty
bool _include_hw =
_include_common && !itr.m_metrics.get_hw_counters().empty();
std::string hw_counter_json = "{}";
if(_include_hw)
{
// current values when read
hw_counter_json = generate_hw_counter_json(_tid, itr.m_metrics);
}
if(get_sampling_include_inlines() && iitr.lineinfo)
{
auto _lines = iitr.lineinfo.lines;
std::reverse(_lines.begin(), _lines.end());
size_t _n = 0;
for(const auto& line : _lines)
{
const auto* _name =
static_strings.emplace(demangle(line.name)).first->c_str();
auto inlined_name_id = data_processor.insert_string(_name);
auto inlined_call_stack = ::rocpd::json::create();
inlined_call_stack->set("name", std::string(demangle(line.name)));
inlined_call_stack->set("location", std::string(line.location));
inlined_call_stack->set("line", std::to_string(line.line));
inlined_call_stack->set("inlined", "true");
rocpd_insert_region<category::timer_sampling>(
thread_primary_key, _beg, _end, inlined_name_id,
_track_name.c_str(), inlined_call_stack->to_string().c_str(),
"{}",
// Only include HW counters for first inlined function
(_n == 0) ? hw_counter_json.c_str() : "{}");
}
}
else
{
const auto* _name = static_strings.emplace(iitr.name).first->c_str();
const auto name_id = data_processor.insert_string(_name);
rocpd_insert_region<category::timer_sampling>(
thread_primary_key, _beg, _end, name_id, _track_name.c_str(),
generate_call_stack_json(iitr).c_str(),
generate_line_info_json(iitr).c_str(), hw_counter_json.c_str());
}
}
}
}
#endif
}
void
post_process_rocpd(int64_t _tid, const std::vector<timer_sampling_data>& _timer_data,
const std::vector<overflow_sampling_data>& _overflow_data)
{
#if ROCPROFSYS_USE_ROCM > 0
rocpd_post_process_overflow_data(_tid, _overflow_data);
rocpd_post_process_timer_data(_tid, _timer_data);
#else
(void) _tid;
(void) _timer_data;
(void) _overflow_data;
cache_sampling_data(_tid, _timer_data, _overflow_data);
cache_backtrace_metrics(_tid, _timer_data);
#endif
}