SWDEV-488948: PC Sampling - Correlation class to provide some thread safety. Adding multithread tests. (#1112)

* SWDEV-488948: PC Sampling - Correlation class to provide some thread safety. Adding multithread tests. * Update source/lib/rocprofiler-sdk/pc_sampling/parser/correlation.hpp Co-authored-by: Vladimir Indic <139573562+vlaindic@users.noreply.github.com> * Update source/lib/rocprofiler-sdk/pc_sampling/parser/correlation.hpp Co-authored-by: Vladimir Indic <139573562+vlaindic@users.noreply.github.com> * Adding backlog for codeobj changes * Formatting * Update source/lib/rocprofiler-sdk/pc_sampling/code_object.hpp Co-authored-by: Vladimir Indic <139573562+vlaindic@users.noreply.github.com> * Update source/lib/rocprofiler-sdk/pc_sampling/code_object.hpp Co-authored-by: Vladimir Indic <139573562+vlaindic@users.noreply.github.com> --------- Co-authored-by: Vladimir Indic <139573562+vlaindic@users.noreply.github.com> [ROCm/rocprofiler-sdk commit: a6be3f9660]
2024-10-07 17:00:59 -03:00
@@ -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)
@@ -45,13 +45,6 @@ namespace pc_sampling
 {
 namespace code_object
 {
-CodeobjTableTranslatorSynchronized*
-get_code_object_translator()
-{
-    static auto*& _v = common::static_object<CodeobjTableTranslatorSynchronized>::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});
@@ -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 <rocprofiler-sdk/callback_tracing.h>
-#    include <rocprofiler-sdk/cxx/codeobj/segment.hpp>
-
-#    include <hsa/hsa_api_trace.h>
-
-#    include <mutex>
-#    include <shared_mutex>
+#include <hsa/hsa_api_trace.h>
+#include <rocprofiler-sdk/callback_tracing.h>
+#include <mutex>
+#include <rocprofiler-sdk/cxx/codeobj/segment.hpp>
+#include <shared_mutex>

 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<std::shared_mutex> 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<CodeobjTableTranslatorSynchronized>::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<address_range_t> insert_backlog{};
+    std::vector<address_range_t> 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
@@ -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 <rocprofiler-sdk/fwd.h>
-
+#include <atomic>
 #include <cstdint>
 #include <iostream>
 #include <memory>
+#include <mutex>
 #include <unordered_map>
 #include <vector>

@@ -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<size_t> _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<std::mutex> 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<std::mutex> 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<std::mutex> 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<std::mutex> 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<DispatchPkt, rocprofiler_correlation_id_t> dispatch_to_correlation{};
+    std::atomic<size_t>                                           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 <bool bHostTrap, typename GFXIP>
 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<const perf_sample_snapshot_v1*>(buffer + p);
-        samples[p]       = copySample<bHostTrap, GFXIP>((const void*) (buffer + p));

        auto& pc_sample = samples[p];
+        pc_sample       = copySample<bHostTrap, GFXIP>((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;
@@ -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")
@@ -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<MockRuntimeBuffer> buffer = std::make_shared<MockRuntimeBuffer>();
    std::array<std::vector<std::shared_ptr<MockDispatch>>, NUM_QUEUES> active_dispatches;
@@ -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<packet_union_t>                                packets;
    std::vector<std::vector<rocprofiler_pc_sampling_record_t>> 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<size_t> reset_available_ids()
    {
        std::unordered_set<size_t> set;
@@ -122,9 +130,18 @@ private:
            set.insert(i << 3);
        return set;
    };
-    static std::unordered_set<size_t> available_ids;
+
+    static std::unique_lock<std::mutex> getLock()
+    {
+        static std::mutex mut;
+        return std::unique_lock<std::mutex>(mut);
+    }
+    static std::unordered_set<size_t>& getAvailableIds()
+    {
+        static std::unordered_set<size_t> available_ids = reset_available_ids();
+        return available_ids;
+    }
 };
-std::unordered_set<size_t> 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<MockRuntimeBuffer>& 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<int> 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<MockRuntimeBuffer> const buffer;

 private:
-    static size_t cur_unique_id;
+    static size_t getUniqueId()
+    {
+        static std::atomic<size_t> _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<MockQueue> 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<size_t> _id{1};
+        return _id.fetch_add(1);
+    }
 };
-size_t MockDispatch::cur_unique_id = 0;

 /**
 * Lightweight class to represent a wave in the particular dispatch.
@@ -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 <gtest/gtest.h>
+#include <cstddef>
+
+#include <future>
+#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<size_t> counter;
+};
+
+/**
+ * Sample user memory allocation callback.
+ * It expects userdata to be cast-able to a pointer to
+ * std::vector<std::pair<rocprofiler_pc_sampling_record_t*, uint64_t>>
+ */
+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<std::vector<std::pair<rocprofiler_pc_sampling_record_t*, uint64_t>>*>(
+            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<MockRuntimeBuffer> buffer = std::make_shared<MockRuntimeBuffer>(tid);
+
+    std::array<std::shared_ptr<MockQueue>, NUM_QUEUES>                 queues;
+    std::array<std::vector<std::shared_ptr<MockDispatch>>, 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<MockQueue>(QSIZE, buffer);
+    for(int i = 0; i < NUM_QUEUES; i++)
+        active_dispatches[i].push_back(std::make_shared<MockDispatch>(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<MockQueue>(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<MockDispatch>(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<MockDispatch> 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<std::pair<rocprofiler_pc_sampling_record_t*, uint64_t>> all_allocations;
+
+    CHECK_PARSER(_parse_buffer<GFX9>((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<size_t, size_t>
+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<MockRuntimeBuffer> buffer = std::make_shared<MockRuntimeBuffer>(tid);
+    std::array<std::vector<std::shared_ptr<MockDispatch>>, NUM_QUEUES> active_dispatches;
+
+    for(size_t q = 0; q < NUM_QUEUES; q++)
+    {
+        std::shared_ptr<MockQueue> queue = std::make_shared<MockQueue>(DISP_PER_QUEUE * 2, buffer);
+        for(size_t d = 0; d < DISP_PER_QUEUE; d++)
+            active_dispatches[q].push_back(std::make_shared<MockDispatch>(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<rocprofiler_pc_sampling_record_t*, size_t> 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<GFX9>(
+        (generic_sample_t*) buffer->packets.data(),
+        buffer->packets.size(),
+        [](rocprofiler_pc_sampling_record_t** sample, uint64_t size, void* userdata_) {
+            auto* pair =
+                reinterpret_cast<std::pair<rocprofiler_pc_sampling_record_t*, size_t>*>(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<MockRuntimeBuffer>(tid);
+    auto queue  = std::make_shared<MockQueue>(QSIZE, buffer);
+
+    std::pair<rocprofiler_pc_sampling_record_t*, size_t> 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<MockDispatch>(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<GFX9>(
+            (generic_sample_t*) buffer->packets.data(),
+            buffer->packets.size(),
+            [](rocprofiler_pc_sampling_record_t** sample, uint64_t size, void* userdata_) {
+                auto* pair =
+                    reinterpret_cast<std::pair<rocprofiler_pc_sampling_record_t*, size_t>*>(
+                        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<std::future<std::pair<size_t, size_t>>> 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<std::future<void>> 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<std::future<void>> threads{};
+    for(size_t i = 0; i < NUM_THREADS; i++)
+        threads.push_back(std::async(std::launch::async, multithread_codeobj, i, &latch));
+};