diff --git a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/pc_sampling/CMakeLists.txt b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/pc_sampling/CMakeLists.txt index 468ffcc4d2..415b934746 100644 --- a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/pc_sampling/CMakeLists.txt +++ b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/pc_sampling/CMakeLists.txt @@ -2,6 +2,8 @@ # PC Sampling # - HSA support officially added in HSA-Runtime v1.14.0 # +add_subdirectory(parser) + if(hsa-runtime64_VERSION AND hsa-runtime64_VERSION VERSION_LESS 1.14.0) return() endif() @@ -14,7 +16,6 @@ set(ROCPROFILER_PC_SAMPLING_HEADERS hsa_adapter.hpp utils.hpp service.hpp types. target_sources(rocprofiler-object-library PRIVATE ${ROCPROFILER_PC_SAMPLING_SOURCES} ${ROCPROFILER_PC_SAMPLING_HEADERS}) -add_subdirectory(parser) add_subdirectory(ioctl) if(ROCPROFILER_BUILD_TESTS) diff --git a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/pc_sampling/code_object.cpp b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/pc_sampling/code_object.cpp index adbef555ce..eb15530394 100644 --- a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/pc_sampling/code_object.cpp +++ b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/pc_sampling/code_object.cpp @@ -45,13 +45,6 @@ namespace pc_sampling { namespace code_object { -CodeobjTableTranslatorSynchronized* -get_code_object_translator() -{ - static auto*& _v = common::static_object::construct(); - return _v; -} - namespace { auto& @@ -110,7 +103,7 @@ executable_freeze(hsa_executable_t executable, const char* options) if(code_object.hsa_executable == executable) { const auto& code_object_rocp = code_object.rocp_data; - get_code_object_translator()->insert( + CodeobjTableTranslatorSynchronized::Get()->insert( address_range_t{code_object_rocp.load_base, code_object_rocp.load_size, code_object_rocp.code_object_id}); @@ -130,7 +123,7 @@ executable_destroy(hsa_executable_t executable) { flush_pc_sampling_buffers(code_object); const auto& code_object_rocp = code_object.rocp_data; - get_code_object_translator()->remove( + CodeobjTableTranslatorSynchronized::Get()->remove( address_range_t{code_object_rocp.load_base, code_object_rocp.load_size, code_object_rocp.code_object_id}); diff --git a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/pc_sampling/code_object.hpp b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/pc_sampling/code_object.hpp index 1ab87af58a..739a8da6d4 100644 --- a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/pc_sampling/code_object.hpp +++ b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/pc_sampling/code_object.hpp @@ -25,15 +25,11 @@ #include "lib/common/static_object.hpp" #include "lib/rocprofiler-sdk/pc_sampling/defines.hpp" -#if ROCPROFILER_SDK_HSA_PC_SAMPLING > 0 - -# include -# include - -# include - -# include -# include +#include +#include +#include +#include +#include namespace rocprofiler { @@ -41,12 +37,6 @@ namespace pc_sampling { namespace code_object { -void -initialize(HsaApiTable* table); - -void -finalize(); - using address_range_t = rocprofiler::sdk::codeobj::segment::address_range_t; using CodeobjTableTranslator = rocprofiler::sdk::codeobj::segment::CodeobjTableTranslator; @@ -59,38 +49,82 @@ public: // Must acquire write lock void insert(address_range_t addr_range) { - auto lock = std::unique_lock{mut}; - this->Super::insert(addr_range); + auto lock = std::unique_lock{backlog_mut}; + insert_backlog.emplace_back(addr_range); + + if(auto try_lock = std::unique_lock{query_mut, std::try_to_lock}) clear_insert_log(); } // Must acquire write lock - bool remove(address_range_t addr_range) + void remove(address_range_t addr_range) { - auto lock = std::unique_lock{mut}; - return this->Super::remove(addr_range); + auto lock = std::unique_lock{backlog_mut}; + remove_backlog.emplace_back(addr_range); + + if(auto try_lock = std::unique_lock{query_mut, std::try_to_lock}) clear_remove_log(); } + void clear_backlog() + { + auto backlog_lock = std::unique_lock{backlog_mut}; + + if(!remove_backlog.empty() || !insert_backlog.empty()) + { + auto query_lock = std::unique_lock{query_mut}; + + clear_remove_log(); + clear_insert_log(); + } + } + + std::shared_lock acquire_query_lock() { return std::shared_lock{query_mut}; } + // Must acquire read lock address_range_t find_codeobj_in_range(uint64_t addr) const { - // TODO: It would be good to have a way to cache search results - // (caching could be done easily in the parser) - auto lock = std::shared_lock{mut}; - auto it = this->find(address_range_t{addr, 0, 0}); + auto it = this->find(address_range_t{addr, 0, 0}); // `addr` might originate from an unknown code object. if(it == this->end()) return address_range_t{0, 0, ROCPROFILER_CODE_OBJECT_ID_NONE}; return *it; } + static CodeobjTableTranslatorSynchronized* Get() + { + static auto*& _v = common::static_object::construct(); + return _v; + } + private: - mutable std::shared_mutex mut = {}; + void clear_insert_log() + { + for(const auto& addr_range : insert_backlog) + this->Super::insert(addr_range); + insert_backlog.clear(); + } + + void clear_remove_log() + { + for(const auto& addr_range : remove_backlog) + this->Super::remove(addr_range); + remove_backlog.clear(); + } + + std::mutex backlog_mut{}; + std::shared_mutex query_mut{}; + std::vector insert_backlog{}; + std::vector remove_backlog{}; }; -CodeobjTableTranslatorSynchronized* -get_code_object_translator(); +#if ROCPROFILER_SDK_HSA_PC_SAMPLING > 0 + +void +initialize(HsaApiTable* table); + +void +finalize(); + +#endif } // namespace code_object } // namespace pc_sampling } // namespace rocprofiler - -#endif diff --git a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/pc_sampling/parser/correlation.hpp b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/pc_sampling/parser/correlation.hpp index ec6e73ea84..8aa22192f9 100644 --- a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/pc_sampling/parser/correlation.hpp +++ b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/pc_sampling/parser/correlation.hpp @@ -22,15 +22,15 @@ #pragma once -#include "lib/common/logging.hpp" +#include "include/rocprofiler-sdk/cxx/codeobj/code_printing.hpp" #include "lib/rocprofiler-sdk/pc_sampling/code_object.hpp" #include "lib/rocprofiler-sdk/pc_sampling/parser/translation.hpp" -#include - +#include #include #include #include +#include #include #include @@ -68,6 +68,15 @@ struct DispatchPkt device_handle dev; //! Which device this is run }; +struct cache_type_t +{ + trap_correlation_id_t id_in{.raw = ~0ul}; + rocprofiler_correlation_id_t id_out{}; + uint64_t dev_id = ~0ul; + size_t increment = 0; + size_t object_id = 0; +}; + inline bool operator==(const trap_correlation_id_t& a, const trap_correlation_id_t& b) { @@ -99,7 +108,11 @@ namespace Parser class CorrelationMap { public: - CorrelationMap() = default; + CorrelationMap() + { + static std::atomic _ids{1}; + object_id = _ids.fetch_add(1); + }; /** * Checks wether a dispatch pkt will generate a collision. @@ -116,9 +129,10 @@ public: */ void newDispatch(const dispatch_pkt_id_t& pkt) { - cache_dev_id = ~0ul; + std::unique_lock lk(mut); auto trap_id = trap_correlation_id(pkt.doorbell_id, pkt.write_index, pkt.queue_size); dispatch_to_correlation[{trap_id, pkt.device}] = pkt.correlation_id; + cache_reset_count.fetch_add(1); } /** @@ -126,9 +140,10 @@ public: */ void forget(const dispatch_pkt_id_t& pkt) { - cache_dev_id = ~0ul; + std::unique_lock lk(mut); auto trap_id = trap_correlation_id(pkt.doorbell_id, pkt.write_index, pkt.queue_size); dispatch_to_correlation.erase({trap_id, pkt.device}); + cache_reset_count.fetch_add(1); } /** @@ -138,13 +153,25 @@ public: rocprofiler_correlation_id_t get(device_handle dev, trap_correlation_id_t correlation_in) { #ifndef _PARSER_CORRELATION_DISABLE_CACHE - if(dev.handle == cache_dev_id && correlation_in == cache_correlation_id_in) - return cache_correlation_id_out; + static thread_local cache_type_t cache{}; + size_t new_increment = cache_reset_count.load(); + + if(cache.increment == new_increment && cache.object_id == this->object_id && + cache.dev_id == dev.handle && cache.id_in == correlation_in) + return cache.id_out; + + // Using unique_lock showed better performance over the shared_lock + std::unique_lock lk(mut); + cache.increment = cache_reset_count.load(); + cache.object_id = object_id; + cache.id_out = dispatch_to_correlation.at({correlation_in, dev}); + cache.dev_id = dev.handle; + cache.id_in = correlation_in; + return cache.id_out; +#else + std::unique_lock lk(mut); + return dispatch_to_correlation.at({correlation_in, dev}); #endif - cache_correlation_id_out = dispatch_to_correlation.at({correlation_in, dev}); - cache_dev_id = dev.handle; - cache_correlation_id_in = correlation_in; - return cache_correlation_id_out; } /** @@ -169,16 +196,15 @@ public: private: std::unordered_map dispatch_to_correlation{}; + std::atomic cache_reset_count{1}; + size_t object_id = 0; - // Making get() const and these cache variables mutable causes performance to be unstable - trap_correlation_id_t cache_correlation_id_in{.raw = ~0ul}; // Invalid value in cache - rocprofiler_correlation_id_t cache_correlation_id_out{ - .internal = ~0ul, - .external = rocprofiler_user_data_t{.value = ~0ul}}; - uint64_t cache_dev_id = ~0ul; // Invalid device Id in cache + std::mutex mut; }; } // namespace Parser +using address_range_t = rocprofiler::sdk::codeobj::segment::address_range_t; + template inline pcsample_status_t add_upcoming_samples(const device_handle device, @@ -187,23 +213,25 @@ add_upcoming_samples(const device_handle device, Parser::CorrelationMap* corr_map, rocprofiler_pc_sampling_record_t* samples) { - pcsample_status_t status = PCSAMPLE_STATUS_SUCCESS; - auto cache_addr_range = rocprofiler::pc_sampling::code_object::address_range_t{0, 0, 0}; - const auto* code_object_translator = - rocprofiler::pc_sampling::code_object::get_code_object_translator(); + pcsample_status_t status = PCSAMPLE_STATUS_SUCCESS; + auto cache_addr_range = address_range_t{0, 0, ROCPROFILER_CODE_OBJECT_ID_NONE}; + + auto* table = rocprofiler::pc_sampling::code_object::CodeobjTableTranslatorSynchronized::Get(); + // To achieve better performance, we exported mutex outside of the translator class. + table->clear_backlog(); + auto table_read_lock = table->acquire_query_lock(); + for(uint64_t p = 0; p < available_samples; p++) { const auto* snap = reinterpret_cast(buffer + p); - samples[p] = copySample((const void*) (buffer + p)); auto& pc_sample = samples[p]; + pc_sample = copySample((const void*) (buffer + p)); pc_sample.size = sizeof(rocprofiler_pc_sampling_record_t); // Convert PC -> (loaded code object id containing PC, offset within code object) if(!cache_addr_range.inrange(snap->pc)) - { - cache_addr_range = code_object_translator->find_codeobj_in_range(snap->pc); - } + cache_addr_range = table->find_codeobj_in_range(snap->pc); pc_sample.pc.loaded_code_object_id = cache_addr_range.id; pc_sample.pc.loaded_code_object_offset = snap->pc - cache_addr_range.addr; @@ -232,8 +260,7 @@ _parse_buffer(generic_sample_t* buffer, Parser::CorrelationMap* corr_map) { // Maximum size - uint64_t index = 0; - + uint64_t index = 0; pcsample_status_t status = PCSAMPLE_STATUS_SUCCESS; while(index < buffer_size) @@ -282,10 +309,7 @@ _parse_buffer(generic_sample_t* buffer, } break; } - default: - ROCP_WARNING << "Index " << index - << " - Invalid sample type: " << buffer[index].type << "\n"; - return PCSAMPLE_STATUS_INVALID_SAMPLE; + default: return PCSAMPLE_STATUS_INVALID_SAMPLE; } } return status; diff --git a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/pc_sampling/parser/tests/CMakeLists.txt b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/pc_sampling/parser/tests/CMakeLists.txt index 0f14442f2a..c8ceca61ca 100644 --- a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/pc_sampling/parser/tests/CMakeLists.txt +++ b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/pc_sampling/parser/tests/CMakeLists.txt @@ -58,3 +58,22 @@ target_link_libraries( pcs_bench_test PRIVATE rocprofiler-sdk::rocprofiler-common-library rocprofiler-sdk::rocprofiler-static-library GTest::gtest GTest::gtest_main) + +add_executable(pcs_thread_test) +target_compile_options(pcs_thread_test PRIVATE "-Ofast") + +target_sources(pcs_thread_test PRIVATE multigpu.cpp) +target_include_directories(pcs_thread_test PRIVATE ${PCTEST_INCLUDE_DIR}) + +target_link_libraries( + pcs_thread_test + PRIVATE rocprofiler-sdk::rocprofiler-common-library + rocprofiler-sdk::rocprofiler-static-library GTest::gtest GTest::gtest_main) + +gtest_add_tests( + TARGET pcs_thread_test + SOURCES multigpu.cpp + TEST_LIST pcs_thread_test_TESTS + WORKING_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}) + +set_tests_properties(${pcs_thread_test_TESTS} PROPERTIES TIMEOUT 75 LABELS "unittests") diff --git a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/pc_sampling/parser/tests/benchmark_test.cpp b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/pc_sampling/parser/tests/benchmark_test.cpp index 5f46a94f7d..4c9248998a 100644 --- a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/pc_sampling/parser/tests/benchmark_test.cpp +++ b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/pc_sampling/parser/tests/benchmark_test.cpp @@ -36,7 +36,7 @@ Benchmark(bool bWarmup) { constexpr size_t SAMPLE_PER_DISPATCH = 8192; constexpr size_t DISP_PER_QUEUE = 8; - constexpr size_t NUM_QUEUES = MockDoorBell::num_unique_bells; + constexpr size_t NUM_QUEUES = 4; std::shared_ptr buffer = std::make_shared(); std::array>, NUM_QUEUES> active_dispatches; diff --git a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/pc_sampling/parser/tests/mocks.hpp b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/pc_sampling/parser/tests/mocks.hpp index 26e6f8e78c..ce56b2ac2a 100644 --- a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/pc_sampling/parser/tests/mocks.hpp +++ b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/pc_sampling/parser/tests/mocks.hpp @@ -49,7 +49,11 @@ class MockRuntimeBuffer { public: - MockRuntimeBuffer() { packets = {}; }; + MockRuntimeBuffer(uint32_t device_ = 0) + : device(device_) + { + packets = {}; + }; //! Adds a packet to the buffer void submit(const packet_union_t& packet) { packets.push_back(packet); }; @@ -62,6 +66,7 @@ public: uni.upcoming.type = AMD_UPCOMING_SAMPLES; uni.upcoming.which_sample_type = AMD_SNAPSHOT_V1; uni.upcoming.num_samples = num_samples; + uni.upcoming.device.handle = device; submit(uni); } @@ -90,6 +95,8 @@ public: std::vector packets; std::vector> parsed_data; + + const uint32_t device; }; /** @@ -100,21 +107,22 @@ class MockDoorBell { public: MockDoorBell() - : handler(getUniqueId()) { - available_ids.erase(handler); + auto lock = getLock(); + assert(getAvailableIds().size() > 0); + handler = *getAvailableIds().begin(); + getAvailableIds().erase(handler); }; - ~MockDoorBell() { available_ids.insert(handler); } + ~MockDoorBell() + { + auto lock = getLock(); + getAvailableIds().insert(handler); + } - const size_t handler; - static constexpr size_t num_unique_bells = 4; + size_t handler; + static constexpr size_t num_unique_bells = 32; private: - static size_t getUniqueId() - { - assert(available_ids.size() > 0); - return *available_ids.begin(); - } static std::unordered_set reset_available_ids() { std::unordered_set set; @@ -122,9 +130,18 @@ private: set.insert(i << 3); return set; }; - static std::unordered_set available_ids; + + static std::unique_lock getLock() + { + static std::mutex mut; + return std::unique_lock(mut); + } + static std::unordered_set& getAvailableIds() + { + static std::unordered_set available_ids = reset_available_ids(); + return available_ids; + } }; -std::unordered_set MockDoorBell::available_ids = MockDoorBell::reset_available_ids(); /** * Mimics a HSA queue. Every live instance of this class has an unique ID and a doorbell. @@ -136,9 +153,10 @@ class MockQueue { public: MockQueue(int size_, std::shared_ptr& buffer_) - : id(cur_unique_id) + : id(getUniqueId()) , size(size_) , doorbell() + , device(buffer_->device) , buffer(buffer_){}; //! Submits a packet to the runtime buffer @@ -153,22 +171,28 @@ public: read_index++; } - int read_index = 0; - int write_index = 0; - size_t active_dispatches = - 0; //! Number of dispatches that are still able to generate PC samples - int last_known_read_pkt = 0; + int read_index = 0; + int write_index = 0; + + size_t active_dispatches = 0; //! Number of dispatches that are still generating PC samples + int last_known_read_pkt = 0; + std::unordered_set async_read_index{}; - const size_t id; - const size_t size; - const MockDoorBell doorbell; + const size_t id; + const size_t size; + const MockDoorBell doorbell; + const uint32_t device; + std::shared_ptr const buffer; private: - static size_t cur_unique_id; + static size_t getUniqueId() + { + static std::atomic _id{1}; + return _id.fetch_add(1); + } }; -size_t MockQueue::cur_unique_id = 1; /** * Mimics a kernel dispatch. @@ -181,12 +205,11 @@ public: : queue(queue_) , dispatch_id(queue->write_index) , doorbell_id(queue->doorbell.handler) - , unique_id(cur_unique_id) + , unique_id(getUniqueId()) { // Ensure queues are not holding more dispatches than queue_size. assert(queue->active_dispatches < queue->size); queue->active_dispatches++; - cur_unique_id++; packet_union_t uni; ::memset(&uni, 0, sizeof(uni)); @@ -195,6 +218,7 @@ public: uni.dispatch_id.queue_size = queue->size; uni.dispatch_id.write_index = dispatch_id; uni.dispatch_id.read_index = queue->read_index; + uni.dispatch_id.device.handle = queue->device; uni.dispatch_id.correlation_id.internal = unique_id; queue->submit(uni); queue->write_index++; @@ -228,15 +252,20 @@ public: } std::shared_ptr const queue; - const size_t dispatch_id; - const size_t doorbell_id; - const size_t unique_id; - static size_t cur_unique_id; + + const size_t dispatch_id; + const size_t doorbell_id; + const size_t unique_id; private: bool queue_read_inc = false; + + static size_t getUniqueId() + { + static std::atomic _id{1}; + return _id.fetch_add(1); + } }; -size_t MockDispatch::cur_unique_id = 0; /** * Lightweight class to represent a wave in the particular dispatch. diff --git a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/pc_sampling/parser/tests/multigpu.cpp b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/pc_sampling/parser/tests/multigpu.cpp new file mode 100644 index 0000000000..21b50b66b0 --- /dev/null +++ b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/pc_sampling/parser/tests/multigpu.cpp @@ -0,0 +1,317 @@ +// MIT License +// +// Copyright (c) 2023 Advanced Micro Devices, Inc. All rights reserved. +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in +// all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +// THE SOFTWARE. + +#include +#include + +#include +#include "lib/rocprofiler-sdk/pc_sampling/code_object.hpp" +#include "mocks.hpp" + +#define GFXIP_MAJOR 9 +constexpr size_t NUM_THREADS = 8; + +class Latch +{ +public: + Latch(size_t num) { counter.store(num); }; + void sync() + { + counter.fetch_sub(1); + while(counter.load()) + ; + }; + std::atomic counter; +}; + +/** + * Sample user memory allocation callback. + * It expects userdata to be cast-able to a pointer to + * std::vector> + */ +static uint64_t +alloc_callback(rocprofiler_pc_sampling_record_t** buffer, uint64_t size, void* userdata) +{ + *buffer = new rocprofiler_pc_sampling_record_t[size]; + auto& vector = + *reinterpret_cast>*>( + userdata); + vector.push_back({*buffer, size}); + return size; +} + +void +multithread_queue_hammer(size_t tid, Latch* latch) +{ + static auto corr_map = Parser::CorrelationMap{}; + std::mt19937 rdgen(tid); + + constexpr int NUM_ACTIONS = 100000; + constexpr int QSIZE = 16; + constexpr int NUM_QUEUES = MockDoorBell::num_unique_bells / NUM_THREADS; + constexpr int ACTION_MAX = QSIZE * NUM_QUEUES / 2; + + std::shared_ptr buffer = std::make_shared(tid); + + std::array, NUM_QUEUES> queues; + std::array>, NUM_QUEUES> active_dispatches; + + int num_reset_queues = 0; + int num_samples_generated = 0; + int num_dispatches_generated = 0; + double avg_q_occupancy = 0; + size_t max_q_occupancy = 0; + + for(int i = 0; i < NUM_QUEUES; i++) + queues[i] = std::make_shared(QSIZE, buffer); + for(int i = 0; i < NUM_QUEUES; i++) + active_dispatches[i].push_back(std::make_shared(queues[i])); + + for(int i = 0; i < NUM_ACTIONS; i++) + { + int q = rdgen() % NUM_QUEUES; + int action = rdgen() % ACTION_MAX; + if(action == 0) + { + // Delete queue and create new one + active_dispatches[q] = {}; + queues[q].reset(); + queues[q] = std::make_shared(QSIZE, buffer); + num_reset_queues++; + } + else if(action > ACTION_MAX / 2 && active_dispatches[q].size() > 1) + { + // Delete dispatch + active_dispatches[q].erase(active_dispatches[q].begin(), + active_dispatches[q].begin() + 1); + } + + // Add new dispatch + if(active_dispatches[q].size() < QSIZE) + { + active_dispatches[q].push_back(std::make_shared(queues[q])); + num_dispatches_generated += 1; + } + + // Generate one "pc" sample for each queue + buffer->genUpcomingSamples(NUM_QUEUES); + for(auto& queue : active_dispatches) + { + EXPECT_NE(queue.size(), 0); + std::shared_ptr rand_dispatch = queue[rdgen() % queue.size()]; + MockWave(rand_dispatch).genPCSample(); + num_samples_generated += 1; + avg_q_occupancy += queue.size(); + max_q_occupancy = std::max(max_q_occupancy, queue.size()); + } + } + + latch->sync(); + + std::vector> all_allocations; + + CHECK_PARSER(_parse_buffer((generic_sample_t*) buffer->packets.data(), + buffer->packets.size(), + alloc_callback, + (void*) &all_allocations, + &corr_map)); + + EXPECT_EQ(all_allocations.size(), NUM_ACTIONS); // Incorrect number of callbacks + for(auto sb = 0ul; sb < all_allocations.size(); sb++) + { + rocprofiler_pc_sampling_record_t* samples = all_allocations[sb].first; + size_t num_samples = all_allocations[sb].second; + + EXPECT_EQ(num_samples, NUM_QUEUES); + for(size_t i = 0; i < num_samples; i++) + EXPECT_EQ(samples[i].correlation_id.internal, samples[i].pc.loaded_code_object_offset); + delete[] samples; + } +} + +/** + * Benchmarks how fast the parser can process samples on a single threaded case + * Current: 5600X with -Ofast, up to >140 million samples/s or ~9GB/s R/W (18GB/s bidirectional) + */ +static std::pair +MultiThread_BenchMark(size_t tid, Latch* latch) +{ + static auto corr_map = Parser::CorrelationMap{}; + + constexpr size_t SAMPLE_PER_DISPATCH = 4096; + constexpr size_t DISP_PER_QUEUE = 16; + constexpr size_t NUM_QUEUES = 1; + + std::shared_ptr buffer = std::make_shared(tid); + std::array>, NUM_QUEUES> active_dispatches; + + for(size_t q = 0; q < NUM_QUEUES; q++) + { + std::shared_ptr queue = std::make_shared(DISP_PER_QUEUE * 2, buffer); + for(size_t d = 0; d < DISP_PER_QUEUE; d++) + active_dispatches[q].push_back(std::make_shared(queue)); + } + + constexpr size_t TOTAL_NUM_SAMPLES = NUM_QUEUES * DISP_PER_QUEUE * SAMPLE_PER_DISPATCH; + buffer->genUpcomingSamples(TOTAL_NUM_SAMPLES); + + for(auto& queue : active_dispatches) + for(auto& dispatch : queue) + for(size_t i = 0; i < SAMPLE_PER_DISPATCH; i++) + MockWave(dispatch).genPCSample(); + + std::pair userdata; + userdata.first = new rocprofiler_pc_sampling_record_t[TOTAL_NUM_SAMPLES]; + userdata.second = TOTAL_NUM_SAMPLES; + + latch->sync(); + + auto t0 = std::chrono::system_clock::now(); + CHECK_PARSER(_parse_buffer( + (generic_sample_t*) buffer->packets.data(), + buffer->packets.size(), + [](rocprofiler_pc_sampling_record_t** sample, uint64_t size, void* userdata_) { + auto* pair = + reinterpret_cast*>(userdata_); + *sample = pair->first; + return size; + }, + &userdata, + &corr_map)); + auto t1 = std::chrono::system_clock::now(); + delete[] userdata.first; + return {TOTAL_NUM_SAMPLES, (t1 - t0).count()}; +} + +void +multithread_codeobj(size_t tid, Latch* latch) +{ + using addr_range_t = rocprofiler::sdk::codeobj::segment::address_range_t; + auto* table = rocprofiler::pc_sampling::code_object::CodeobjTableTranslatorSynchronized::Get(); + + static auto corr_map = Parser::CorrelationMap{}; + std::mt19937 rdgen(tid); + + constexpr int NUM_DISPATCH = 20000; + constexpr int NUM_SAMPLES = 50; + constexpr int QSIZE = 16; + + auto buffer = std::make_shared(tid); + auto queue = std::make_shared(QSIZE, buffer); + + std::pair userdata; + userdata.first = new rocprofiler_pc_sampling_record_t[NUM_SAMPLES]; + userdata.second = NUM_SAMPLES; + + latch->sync(); + + for(int d = 0; d < NUM_DISPATCH; d++) + { + buffer->packets.clear(); + auto dispatch = std::make_shared(queue); + + const size_t pc_base_addr = NUM_SAMPLES * dispatch->unique_id; + table->insert(addr_range_t{pc_base_addr, NUM_SAMPLES, dispatch->unique_id}); + + packet_union_t uni{}; + uni.snap.correlation_id = dispatch->getMockId().raw; + + buffer->genUpcomingSamples(NUM_SAMPLES); + for(int s = 0; s < NUM_SAMPLES; s++) + { + uni.snap.pc = pc_base_addr + s; + dispatch->submit(uni); + } + + CHECK_PARSER(_parse_buffer( + (generic_sample_t*) buffer->packets.data(), + buffer->packets.size(), + [](rocprofiler_pc_sampling_record_t** sample, uint64_t size, void* userdata_) { + auto* pair = + reinterpret_cast*>( + userdata_); + *sample = pair->first; + assert(size <= NUM_SAMPLES); + return size; + }, + &userdata, + &corr_map)); + + for(int s = 0; s < NUM_SAMPLES; s++) + { + const auto& pc = userdata.first[s].pc; + EXPECT_EQ(pc.loaded_code_object_id, dispatch->unique_id); + EXPECT_EQ(pc.loaded_code_object_offset, s); + } + + table->remove(addr_range_t{pc_base_addr, NUM_SAMPLES, dispatch->unique_id}); + } + + delete[] userdata.first; +} + +TEST(pcs_parser, bench_test) +{ + size_t time = 0; + size_t samples = 0; + + for(int it = 0; it < 4; it++) + { + Latch latch(NUM_THREADS); + + std::vector>> threads{}; + for(size_t t = 0; t < NUM_THREADS; t++) + threads.push_back(std::async(std::launch::async, MultiThread_BenchMark, t, &latch)); + + if(it == 0) continue; // Skip warmup + + for(auto& t : threads) + { + auto result = t.get(); + samples += result.first; + time += result.second; + } + } + + double mean = 1E3 * NUM_THREADS * samples / time; + + std::cout << "Benchmark: Parsed " << int(mean * 1E3 + 0.5) * 1E-3f << " Msample/s ("; + std::cout << int(sizeof(rocprofiler_pc_sampling_record_t) * mean) << " MB/s)" << std::endl; +}; + +TEST(pcs_parser, hammer_test) +{ + Latch latch(NUM_THREADS); + + std::vector> threads{}; + for(size_t i = 0; i < NUM_THREADS; i++) + threads.push_back(std::async(std::launch::async, multithread_queue_hammer, i, &latch)); +}; + +TEST(pcs_parser, codeobj_test) +{ + Latch latch(NUM_THREADS); + + std::vector> threads{}; + for(size_t i = 0; i < NUM_THREADS; i++) + threads.push_back(std::async(std::launch::async, multithread_codeobj, i, &latch)); +};