[rocprof-sys] Refactor trace_cache architecture with improved type erasure and processing pipeline (#1710)

- Redesigned buffer_storage with a flush_worker pattern for better thread management and resource cleanup
- Introduced type-safe abstractions through new components: cacheable.hpp, cache_type_traits.hpp, sample_processor.hpp, and type_registry.hpp
- Optimized type erasure implementation in sample processor to reduce runtime overhead
- Renamed rocpd_post_processing to rocpd_processor and restructured the processing pipeline
- Removed storage_parser.cpp and integrated functionality into header-based template implementation
- Enhanced cache_manager with improved processing workflow and better separation of concerns

projects/rocprofiler-systems/source/lib/rocprof-sys/library.cpp

@@ -45,7 +45,7 @@
 #include "core/rocpd/data_processor.hpp"
 #include "core/timemory.hpp"
 #include "core/trace_cache/cache_manager.hpp"
-#include "core/trace_cache/cache_utility.hpp"
+#include "core/trace_cache/cacheable.hpp"
 #include "core/trace_cache/metadata_registry.hpp"
 #include "core/utility.hpp"
 #include "library/causal/data.hpp"
@@ -578,7 +578,7 @@ rocprofsys_init_tooling_hidden(void)
         ROCPROFSYS_DEBUG_F("State: %s -> State::Active\n",
                            std::to_string(get_state()).c_str());
 
-        trace_cache::get_buffer_storage().start_flushing_thread(getpid());
+        trace_cache::get_buffer_storage().start(getpid());
         set_state(State::Active);  // set to active as very last operation
     } };
 
@@ -798,7 +798,7 @@ rocprofsys_finalize_hidden(void)
         const auto _agents  = get_agent_manager_instance().get_agents();
         _manager.shutdown();
         const auto metadata_filepath =
-            trace_cache::get_metadata_filepath(get_root_process_id(), getpid());
+            trace_cache::utility::get_metadata_filepath(get_root_process_id(), getpid());
         _manager.get_metadata_registry().save_to_file(metadata_filepath, _agents);
 
         std::quick_exit(EXIT_SUCCESS);

projects/rocprofiler-systems/source/lib/rocprof-sys/library/amd_smi.cpp

@@ -28,7 +28,7 @@
 
 #include "core/agent.hpp"
 #include "core/trace_cache/cache_manager.hpp"
-#include "core/trace_cache/cache_utility.hpp"
+#include "core/trace_cache/cacheable.hpp"
 #include "core/trace_cache/sample_type.hpp"
 #include <amd_smi/amdsmi.h>
 #include <cstdint>
@@ -693,12 +693,11 @@ data::sample(uint32_t _device_id)
         // Store samples if basic metrics are enabled OR if there's advanced metric data
         if(_basic_metrics_enabled || has_data)
         {
-            trace_cache::get_buffer_storage().store(
-                trace_cache::entry_type::amd_smi_sample, serialize_settings(m_dev_id),
-                _device_id, _timestamp, m_busy_perc.gfx_activity,
-                m_busy_perc.umc_activity, m_busy_perc.mm_activity,
-                m_power.current_socket_power, m_temp, m_mem_usage,
-                serialize_gpu_metrics(m_dev_id, metrics, capabilities));
+            trace_cache::get_buffer_storage().store(trace_cache::amd_smi_sample{
+                serialize_settings(m_dev_id), _device_id, _timestamp,
+                m_busy_perc.gfx_activity, m_busy_perc.umc_activity,
+                m_busy_perc.mm_activity, m_power.current_socket_power, m_temp,
+                m_mem_usage, serialize_gpu_metrics(m_dev_id, metrics, capabilities) });
 
             if(has_data) m_gpu_metrics.push_back(metrics);
         }

projects/rocprofiler-systems/source/lib/rocprof-sys/library/components/backtrace_metrics.cpp

@@ -27,7 +27,7 @@
 #include "core/debug.hpp"
 #include "core/perfetto.hpp"
 #include "core/trace_cache/cache_manager.hpp"
-#include "core/trace_cache/cache_utility.hpp"
+#include "core/trace_cache/cacheable.hpp"
 #include "core/trace_cache/metadata_registry.hpp"
 #include "library/components/ensure_storage.hpp"
 #include "library/ptl.hpp"
@@ -261,11 +261,10 @@ cache_backtrace_metrics_events(const uint32_t device_id, uint64_t timestamp_ns,
     const auto* line_info       = "";
 
     auto insert_event_and_sample = [&](const char* _track_name, double _value) {
-        trace_cache::get_buffer_storage().store(
-            trace_cache::entry_type::pmc_event_with_sample, _track_name, timestamp_ns,
-            event_metadata, stack_id, parent_stack_id, correlation_id, call_stack,
-            line_info, device_id, static_cast<uint8_t>(agent_type::CPU), _track_name,
-            _value);
+        trace_cache::get_buffer_storage().store(trace_cache::pmc_event_with_sample{
+            _track_name, timestamp_ns, event_metadata, stack_id, parent_stack_id,
+            correlation_id, call_stack, line_info, device_id,
+            static_cast<uint8_t>(agent_type::CPU), _track_name, _value });
     };
 
     if constexpr(std::is_same_v<Category, category::thread_hardware_counter>)

projects/rocprofiler-systems/source/lib/rocprof-sys/library/components/category_region.hpp

@@ -27,6 +27,7 @@
 #include "core/state.hpp"
 #include "core/timemory.hpp"
 #include "core/trace_cache/cache_manager.hpp"
+#include "core/trace_cache/sample_type.hpp"
 #include "library/causal/data.hpp"
 #include "library/runtime.hpp"
 #include "library/thread_info.hpp"
@@ -55,9 +56,9 @@ cache_region(uint64_t thread_id, const std::string& name, uint64_t start_ts,
     constexpr const char* CALLSTACK         = "";
     constexpr const char* ARGUMENTS         = "";
     rocprofsys::trace_cache::get_buffer_storage().store(
-        rocprofsys::trace_cache::entry_type::region, thread_id, name.c_str(),
-        NO_CORRELATION_ID, NO_CORRELATION_ID, start_ts, end_ts, CALLSTACK, ARGUMENTS,
-        category.c_str());
+        rocprofsys::trace_cache::region_sample{
+            thread_id, name.c_str(), NO_CORRELATION_ID, NO_CORRELATION_ID, start_ts,
+            end_ts, CALLSTACK, ARGUMENTS, category.c_str() });
 }
 
 struct entry_key

projects/rocprofiler-systems/source/lib/rocprof-sys/library/components/comm_data.cpp

@@ -151,12 +151,11 @@ cache_comm_data_events(const uint32_t device_id, int bytes)
     const std::string call_stack      = "{}";
     const std::string line_info       = "{}";
 
-    trace_cache::get_buffer_storage().store(
-        trace_cache::entry_type::pmc_event_with_sample, track_name.c_str(), timestamp_ns,
-        event_metadata.c_str(), stack_id, parent_stack_id, correlation_id,
-        call_stack.c_str(), line_info.c_str(), device_id,
+    trace_cache::get_buffer_storage().store(trace_cache::pmc_event_with_sample{
+        track_name.c_str(), timestamp_ns, event_metadata.c_str(), stack_id,
+        parent_stack_id, correlation_id, call_stack.c_str(), line_info.c_str(), device_id,
         static_cast<uint8_t>(agent_type::CPU), track_name.c_str(),
-        static_cast<double>(value));
+        static_cast<double>(value) });
 }
 
 }  // namespace

projects/rocprofiler-systems/source/lib/rocprof-sys/library/cpu_freq.cpp

@@ -260,14 +260,13 @@ sample()
     auto _freqs  = component::cpu_freq{}.sample();
 
     // user and kernel mode times are in microseconds
-    trace_cache::get_buffer_storage().store(
-        trace_cache::entry_type::cpu_freq_sample, _timestamp, tim::get_page_rss(),
-        tim::get_virt_mem(), _rcache.get_peak_rss(),
+    trace_cache::get_buffer_storage().store(trace_cache::cpu_freq_sample{
+        _timestamp, tim::get_page_rss(), tim::get_virt_mem(), _rcache.get_peak_rss(),
         _rcache.get_num_priority_context_switch() +
             _rcache.get_num_voluntary_context_switch(),
         _rcache.get_num_major_page_faults() + _rcache.get_num_minor_page_faults(),
         _rcache.get_user_mode_time() * 1000, _rcache.get_kernel_mode_time() * 1000,
-        serialize_freqs(_freqs));
+        serialize_freqs(_freqs) });
 
     data.emplace_back(
         _timestamp, tim::get_page_rss(), tim::get_virt_mem(), _rcache.get_peak_rss(),

projects/rocprofiler-systems/source/lib/rocprof-sys/library/kokkosp.cpp

@@ -191,10 +191,10 @@ cache_kokkos_event(const char* name, const char* event_type, const char* target,
     const char*  line_info       = "{}";
 
     rocprofsys::trace_cache::get_buffer_storage().store(
-        rocprofsys::trace_cache::entry_type::in_time_sample,
-        rocprofsys::trait::name<category::kokkos>::value, timestamp_ns,
-        event_metadata.dump().c_str(), stack_id, parent_stack_id, correlation_id,
-        call_stack, line_info);
+        rocprofsys::trace_cache::in_time_sample{
+            rocprofsys::trait::name<category::kokkos>::value, timestamp_ns,
+            event_metadata.dump().c_str(), stack_id, parent_stack_id, correlation_id,
+            call_stack, line_info });
 }
 
 }  // namespace

projects/rocprofiler-systems/source/lib/rocprof-sys/library/rocprofiler-sdk.cpp

@@ -526,7 +526,6 @@ get_mem_alloc_address(
 }
 #endif
 
-// clang-format off
 void
 cache_region(const rocprofiler_callback_tracing_record_t* record,
              const rocprofiler_timestamp_t                start_timestamp,
@@ -538,92 +537,62 @@ cache_region(const rocprofiler_callback_tracing_record_t* record,
         trace_cache::get_metadata_registry().get_callback_tracing_info();
     auto _name = std::string{ callback_tracing_info.at(record->kind, record->operation) };
 
-    trace_cache::get_buffer_storage().store(
-        trace_cache::entry_type::region,
-        record->thread_id,
-        _name.c_str(),
-        record->correlation_id.internal,
-        get_parent_stack_id(record->correlation_id),
-        start_timestamp,
-        end_timestamp,
-        call_stack.c_str(),
-        args_str.c_str(),
-        category.c_str());
+    trace_cache::get_buffer_storage().store(trace_cache::region_sample{
+        record->thread_id, _name.c_str(), record->correlation_id.internal,
+        get_parent_stack_id(record->correlation_id), start_timestamp, end_timestamp,
+        call_stack.c_str(), args_str.c_str(), category.c_str() });
 }
 
 void
-cache_kernel_dispatch(rocprofiler_buffer_tracing_kernel_dispatch_record_t* record, uint64_t stream_handle)
+cache_kernel_dispatch(rocprofiler_buffer_tracing_kernel_dispatch_record_t* record,
+                      uint64_t                                             stream_handle)
 {
-    auto queue_handle  = record->dispatch_info.queue_id.handle;
+    auto queue_handle = record->dispatch_info.queue_id.handle;
 
     trace_cache::get_metadata_registry().add_queue(queue_handle);
     trace_cache::get_metadata_registry().add_stream(stream_handle);
 
-    trace_cache::get_buffer_storage().store(
-        trace_cache::entry_type::kernel_dispatch,
-        record->start_timestamp,
-        record->end_timestamp,
-        record->thread_id,
-        record->dispatch_info.agent_id.handle,
-        record->dispatch_info.kernel_id,
-        record->dispatch_info.dispatch_id,
-        record->dispatch_info.queue_id.handle,
-        record->correlation_id.internal,
-        get_parent_stack_id(record->correlation_id),
+    trace_cache::get_buffer_storage().store(trace_cache::kernel_dispatch_sample{
+        record->start_timestamp, record->end_timestamp, record->thread_id,
+        record->dispatch_info.agent_id.handle, record->dispatch_info.kernel_id,
+        record->dispatch_info.dispatch_id, record->dispatch_info.queue_id.handle,
+        record->correlation_id.internal, get_parent_stack_id(record->correlation_id),
         record->dispatch_info.private_segment_size,
-        record->dispatch_info.group_segment_size,
-        record->dispatch_info.workgroup_size.x,
-        record->dispatch_info.workgroup_size.y,
-        record->dispatch_info.workgroup_size.z,
-        record->dispatch_info.grid_size.x,
-        record->dispatch_info.grid_size.y,
-        record->dispatch_info.grid_size.z,
-        stream_handle);
-
+        record->dispatch_info.group_segment_size, record->dispatch_info.workgroup_size.x,
+        record->dispatch_info.workgroup_size.y, record->dispatch_info.workgroup_size.z,
+        record->dispatch_info.grid_size.x, record->dispatch_info.grid_size.y,
+        record->dispatch_info.grid_size.z, stream_handle });
 }
 
 void
-cache_memory_copy(rocprofiler_buffer_tracing_memory_copy_record_t* record, uint64_t stream_handle)
+cache_memory_copy(rocprofiler_buffer_tracing_memory_copy_record_t* record,
+                  uint64_t                                         stream_handle)
 {
     trace_cache::get_metadata_registry().add_stream(stream_handle);
-    trace_cache::get_buffer_storage().store(
-        trace_cache::entry_type::memory_copy,
-        record->start_timestamp,
-        record->end_timestamp,
-        record->thread_id,
-        record->dst_agent_id.handle,
-        record->src_agent_id.handle,
-        static_cast<int32_t>(record->kind),
-        static_cast<int32_t>(record->operation),
-        record->bytes,
-        record->correlation_id.internal,
-        get_parent_stack_id(record->correlation_id),
-        get_mem_copy_dst_address(*record),
-        get_mem_copy_src_address(*record),
-        stream_handle);
+    trace_cache::get_buffer_storage().store(trace_cache::memory_copy_sample{
+
+        record->start_timestamp, record->end_timestamp, record->thread_id,
+        record->dst_agent_id.handle, record->src_agent_id.handle,
+        static_cast<int32_t>(record->kind), static_cast<int32_t>(record->operation),
+        record->bytes, record->correlation_id.internal,
+        get_parent_stack_id(record->correlation_id), get_mem_copy_dst_address(*record),
+        get_mem_copy_src_address(*record), stream_handle });
 }
 
-#if (ROCPROFILER_VERSION >= 600)
+#if(ROCPROFILER_VERSION >= 600)
 void
-cache_memory_allocation(rocprofiler_buffer_tracing_memory_allocation_record_t* record, uint64_t stream_handle)
+cache_memory_allocation(rocprofiler_buffer_tracing_memory_allocation_record_t* record,
+                        uint64_t stream_handle)
 {
     trace_cache::get_metadata_registry().add_stream(stream_handle);
-    trace_cache::get_buffer_storage().store(
-        trace_cache::entry_type::memory_alloc,
-        record->start_timestamp,
-        record->end_timestamp,
-        record->thread_id,
-        record->agent_id.handle,
-        static_cast<int32_t>(record->kind),
-        static_cast<int32_t>(record->operation),
-        record->allocation_size,
-        record->correlation_id.internal,
-        get_parent_stack_id(record->correlation_id),
-        get_mem_alloc_address(*record),
-        stream_handle);
+    trace_cache::get_buffer_storage().store(trace_cache::memory_allocate_sample{
+        record->start_timestamp, record->end_timestamp, record->thread_id,
+        record->agent_id.handle, static_cast<int32_t>(record->kind),
+        static_cast<int32_t>(record->operation), record->allocation_size,
+        record->correlation_id.internal, get_parent_stack_id(record->correlation_id),
+        get_mem_alloc_address(*record), stream_handle });
 }
 #endif
-// clang-format on
 
 std::string
 get_args_string(const function_args_t& args)

projects/rocprofiler-systems/source/lib/rocprof-sys/library/rocprofiler-sdk/counters.cpp

@@ -120,12 +120,11 @@ counter_event::operator()(const client_data* tool_data, ::perfetto::CounterTrack
 
         auto agent = get_agent_manager_instance().get_agent_by_handle(agent_handle);
 
-        trace_cache::get_buffer_storage().store(
-            trace_cache::entry_type::pmc_event_with_sample, track_name.c_str(),
-            _timing.start, event_metadata.c_str(), stack_id, parent_stack_id,
-            correlation_id, call_stack.c_str(), line_info.c_str(),
+        trace_cache::get_buffer_storage().store(trace_cache::pmc_event_with_sample{
+            track_name.c_str(), _timing.start, event_metadata.c_str(), stack_id,
+            parent_stack_id, correlation_id, call_stack.c_str(), line_info.c_str(),
             static_cast<uint32_t>(agent.device_id), static_cast<uint8_t>(agent.type),
-            track_name.c_str(), value);
+            track_name.c_str(), static_cast<double>(value) });
     }
 }
 

projects/rocprofiler-systems/source/lib/rocprof-sys/library/sampling.cpp

@@ -315,13 +315,12 @@ cache_sampling_data(int64_t _tid, const std::vector<timer_sampling_data>& _timer
             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,
+            trace_cache::get_buffer_storage().store(trace_cache::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(), "{}");
+                _line_info.c_str(), "{}" });
         }
     }
 
@@ -348,13 +347,12 @@ cache_sampling_data(int64_t _tid, const std::vector<timer_sampling_data>& _timer
             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,
+            trace_cache::get_buffer_storage().store(trace_cache::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(), "{}");
+                _line_info.c_str(), "{}" });
         }
     }
 }