SWDEV-378008: Adding changes to serialize the kernels in rocprofV2

Change-Id: I82353ba94b3a15fdc5991e6129fe47f6765a9f74


[ROCm/rocprofiler commit: 54f6e2afb7]

projects/rocprofiler/src/api/rocprofiler_singleton.cpp

@@ -1,3 +1,4 @@
+
 /* Copyright (c) 2022 Advanced Micro Devices, Inc.
 
  Permission is hereby granted, free of charge, to any person obtaining a copy
@@ -135,8 +136,6 @@ rocprofiler_session_id_t ROCProfiler_Singleton::CreateSession(
 }
 
 void ROCProfiler_Singleton::DestroySession(rocprofiler_session_id_t session_id) {
-  while (GetCurrentActiveInterruptSignalsCount() != 0) {
-  }
   {
     std::lock_guard<std::mutex> lock(session_map_lock_);
     ASSERTM(sessions_.find(session_id.handle) != sessions_.end(),
@@ -145,7 +144,7 @@ void ROCProfiler_Singleton::DestroySession(rocprofiler_session_id_t session_id)
     sessions_.erase(session_id.handle);
   }
 }
-
+profiler_serializer_t& ROCProfiler_Singleton::GetSerializer() { return profiler_serializer; }
 bool ROCProfiler_Singleton::FindDeviceProfilingSession(rocprofiler_session_id_t session_id) {
   std::lock_guard<std::mutex> lock(device_profiling_session_map_lock_);
   return dev_profiling_sessions_.find(session_id.handle) != dev_profiling_sessions_.end();

projects/rocprofiler/src/api/rocprofiler_singleton.h

@@ -42,8 +42,28 @@
 #include "src/core/session/session.h"
 #include "src/core/session/device_profiling.h"
 #include "src/core/hardware/hsa_info.h"
-
+#include "src/core/hsa/queues/queue.h"
 namespace rocprofiler {
+  /*This is a profiler serializer. It should be instantiated
+ only once for the profiler. The following is the
+ description of each field.
+ 1. dispatch_queue - The queue to which the currently dispatched kernel
+            belongs to.
+            At any given time, in serialization only one kernel
+            can be executing.
+ 2. dispatch_ready- It is a software data structure which holds
+            the queues which have a kernel ready to be dispatched.
+            This stores the queues in FIFO order.
+ 3. serializer_mutex - The mutex is used for thread synchronization
+            while accessing the singleton instance of this structure.
+ Currently, in case of profiling kernels are serialized by default.
+*/
+struct profiler_serializer_t {
+  queue::Queue* dispatch_queue{nullptr};
+  std::vector<rocprofiler::queue::Queue*> dispatch_ready;
+  std::mutex serializer_mutex;
+};
+
 
 class ROCProfiler_Singleton {
  public:
@@ -79,7 +99,7 @@ class ROCProfiler_Singleton {
                                                         int cpu_agent_index, int gpu_agent_index);
   void DestroyDeviceProfilingSession(rocprofiler_session_id_t session_id);
   DeviceProfileSession* GetDeviceProfilingSession(rocprofiler_session_id_t session_id);
-
+  profiler_serializer_t& GetSerializer();
 
   // Generic
   bool CheckFilterData(rocprofiler_filter_kind_t filter_kind,
@@ -99,7 +119,7 @@ class ROCProfiler_Singleton {
   std::unordered_map<uint64_t, Agent::DeviceInfo> agent_device_map_;
   ROCProfiler_Singleton();
   ~ROCProfiler_Singleton();
-
+   profiler_serializer_t profiler_serializer;
   /*
    * XXX: Associating PC samples with a running kernel requires an identifier
    * that will be unique across all kernel executions.  It is not enough to use

projects/rocprofiler/src/core/hsa/queues/queue.cpp

@@ -368,6 +368,79 @@ void AddAttRecord(rocprofiler_record_att_tracer_t* record, hsa_agent_t gpu_agent
   record->shader_engine_data_count = data.size();
 }
 
+/*
+    Function name: enable_dispatch
+    Argument : pointer to the the Queue class object
+    Description: This function asserts if the mutex is not already
+                 locked by the calling thread. It enable the kernel dispatch
+                 from the given queue by setting its block signal to 0.
+                 Finally, it updates the serializer queue with the given queue.
+*/
+void enable_dispatch(Queue* dispatch_queue) {
+  // ToDO(srnagara): Find a way to assert if the mutex is already locked.
+  // assert(!rocmtools::GetSerializer()->serializer_mutex.try_lock());
+  profiler_serializer_t& serializer =
+      rocprofiler::ROCProfiler_Singleton::GetInstance().GetSerializer();
+  assert(serializer.dispatch_queue == nullptr);
+  HSASupport_Singleton::GetInstance().GetCoreApiTable().hsa_signal_store_screlease_fn(
+      dispatch_queue->GetBlockSignal(), 0);
+  serializer.dispatch_queue = dispatch_queue;
+}
+
+/*
+    Function name:  AsyncSignalReadyHandler
+    Argument:    hsa signal value for which the async handler was called
+                 and pointer to the data.
+    Description: This async handler is invoked when the queue is ready
+                 to launch a kernel. It first, resets the queue's ready signal to 1.
+                 It then checks if there is any queue which has a kernel currently dispatched.
+                 If yes, it pushes the queue to the dispatch ready else
+                 it enables the dispatch for the given queue.
+    Return :     It returns true since we need this handler to be invoked
+                 each time the queue's ready signal (used for entire queue) is set to 0.
+                 If we had a separate signal for every dispatch in the queue then we don't
+                 need this to be invoked more than once in which case we would return false.
+*/
+
+bool AsyncSignalReadyHandler(hsa_signal_value_t signal_value, void* data) {
+  HSASupport_Singleton& hsasupport_singleton = HSASupport_Singleton::GetInstance();
+  profiler_serializer_t& serializer =
+      rocprofiler::ROCProfiler_Singleton::GetInstance().GetSerializer();
+  std::lock_guard<std::mutex> serializer_lock(serializer.serializer_mutex);
+  auto queue = static_cast<Queue*>(data);
+  std::lock_guard<std::mutex> queue_lock(queue->qw_mutex);
+  /* If is_destroy is set by the destructor then unreg_async_handler is set
+  ready signal is destroyed and
+  the destructor is notified and the handler is unregistered by returning false
+  */
+  if (queue->state == is_destroy::to_destroy) {
+    {
+      queue->state = done_destroy;
+      hsasupport_singleton.GetCoreApiTable().hsa_signal_destroy_fn(queue->GetReadySignal());
+    }
+    queue->cv_ready_signal.notify_one();
+    return false;
+  }
+  hsasupport_singleton.GetCoreApiTable().hsa_signal_store_screlease_fn(queue->GetReadySignal(), 1);
+  if (serializer.dispatch_queue == nullptr)
+    enable_dispatch(queue);
+  else
+    serializer.dispatch_ready.push_back(queue);
+  return true;
+}
+/*
+    Function name:  SignalAsyncReadyHandler.
+    Argument :      The signal value and pointer to the data to
+                    pass to the handler.
+    Description :   Registers a asynchronous callback function
+                    for the ready signal to be invoked when it goes to zero.
+*/
+void SignalAsyncReadyHandler(const hsa_signal_t& signal, void* data) {
+  hsa_status_t status =
+      HSASupport_Singleton::GetInstance().GetAmdExtTable().hsa_amd_signal_async_handler_fn(
+          signal, HSA_SIGNAL_CONDITION_EQ, 0, AsyncSignalReadyHandler, data);
+  if (status != HSA_STATUS_SUCCESS) fatal("hsa_amd_signal_async_handler failed");
+}
 bool AsyncSignalHandler(hsa_signal_value_t signal_value, void* data) {
   auto queue_info_session = static_cast<queue_info_session_t*>(data);
   rocprofiler::ROCProfiler_Singleton& rocprofiler_singleton =
@@ -461,6 +534,23 @@ bool AsyncSignalHandler(hsa_signal_value_t signal_value, void* data) {
         }
         delete pending->context;
       }
+
+      /*
+      Check if the dispatch ready is empty, If so, there is no more
+      dispatches to be launched and we return. Else, dispatch the
+      kernel of the queue in the front of the dispatch_ready.
+      */
+      profiler_serializer_t& serializer =
+          rocprofiler::ROCProfiler_Singleton::GetInstance().GetSerializer();
+      std::lock_guard<std::mutex> serializer_lock(serializer.serializer_mutex);
+      assert(serializer.dispatch_queue != nullptr);
+      hsasupport_singleton.GetCoreApiTable().hsa_signal_store_screlease_fn(
+          queue_info_session->block_signal, 1);
+      serializer.dispatch_queue = nullptr;
+      if (serializer.dispatch_ready.empty()) return false;
+      Queue* queue = serializer.dispatch_ready.front();
+      serializer.dispatch_ready.erase(serializer.dispatch_ready.begin());
+      enable_dispatch(queue);
       if (pending->new_signal.handle)
         hsasupport_singleton.GetCoreApiTable().hsa_signal_destroy_fn(pending->new_signal);
       if (queue_info_session->interrupt_signal.handle)
@@ -542,11 +632,23 @@ bool AsyncSignalHandlerATT(hsa_signal_value_t /* signal */, void* data) {
   return false;
 }
 
-void CreateBarrierPacket(const hsa_signal_t& packet_completion_signal,
-                         std::vector<Packet::packet_t>* transformed_packets) {
-  hsa_barrier_and_packet_t barrier{};
+/*
+    Function name: CreateBarrierPacket.
+    Argument :     The list of transformed packets to add the
+                   barrier packet to. Pointer to the completion signal
+                   and the input signal of the barrier packet to be created.
+    Description :  This packet creates the barrier packet with the given
+                   completion signal and dependency signal. It then adds to
+                   the transformed packets list.
+*/
+void CreateBarrierPacket(std::vector<Packet::packet_t>* transformed_packets,
+                         const hsa_signal_t* packet_dependency_signal,
+                         const hsa_signal_t* packet_completion_signal
+                         ) {
+  hsa_barrier_and_packet_t barrier{0};
   barrier.header = HSA_PACKET_TYPE_BARRIER_AND << HSA_PACKET_HEADER_TYPE;
-  barrier.dep_signal[0] = packet_completion_signal;
+  if (packet_completion_signal != nullptr) barrier.completion_signal = *packet_completion_signal;
+  if (packet_dependency_signal != nullptr) barrier.dep_signal[0] = *packet_dependency_signal;
   void* barrier_ptr = &barrier;
   transformed_packets->emplace_back(*reinterpret_cast<Packet::packet_t*>(barrier_ptr));
 }
@@ -800,15 +902,13 @@ void Queue::WriteInterceptor(const void* packets, uint64_t pkt_count, uint64_t u
        is_pc_sampling_collection_mode) &&
       session) {
     // Getting Queue Data and Information
-    Queue& queue_info = *reinterpret_cast<Queue*>(data);
+    auto& queue_info = *reinterpret_cast<Queue*>(data);
     std::lock_guard<std::mutex> lk(queue_info.qw_mutex);
 
 
     // hsa_ven_amd_aqlprofile_profile_t* profile;
     std::vector<std::pair<rocprofiler::profiling_context_t*, hsa_ven_amd_aqlprofile_profile_t*>>
         profiles;
-
-
     // Searching accross all the packets given during this write
     for (size_t i = 0; i < pkt_count; ++i) {
       auto& original_packet = static_cast<const hsa_barrier_and_packet_t*>(packets)[i];
@@ -833,9 +933,25 @@ void Queue::WriteInterceptor(const void* packets, uint64_t pkt_count, uint64_t u
 
       uint32_t profile_id = 0;
       // do {
+
       std::pair<rocprofiler::profiling_context_t*, hsa_ven_amd_aqlprofile_profile_t*> profile;
       if (profiles.size() > 0 && replay_mode_count > 0) profile = profiles.at(profile_id);
 
+      hsa_signal_t ready_signal = queue_info.GetReadySignal();
+      hsa_signal_t block_signal = queue_info.GetBlockSignal();
+
+      /*
+       Creates a barrier packet with its completion signal as the
+       queue's ready signal.
+       */
+      CreateBarrierPacket(&transformed_packets, nullptr, &ready_signal);
+      /*
+      Creates a barrier packet with queue's blocksignal as its input and
+      completion signal.This will ensure it is no longer 0 so a later barrier
+      packet waiting on it to be 0 will be blocked
+      */
+      CreateBarrierPacket(&transformed_packets, &block_signal, &block_signal);
+
       uint32_t writer_id = WRITER_ID.fetch_add(1, std::memory_order_release);
 
       if (session_data_count > 0 && is_counter_collection_mode && profiles.size() > 0 &&
@@ -847,7 +963,8 @@ void Queue::WriteInterceptor(const void* packets, uint64_t pkt_count, uint64_t u
             << HSA_PACKET_HEADER_TYPE;
         AddVendorSpecificPacket(profile.first->start_packet, &transformed_packets, dummy_signal);
 
-        CreateBarrierPacket(profile.first->start_packet->completion_signal, &transformed_packets);
+        CreateBarrierPacket(&transformed_packets, &profile.first->start_packet->completion_signal,
+                            nullptr);
       }
 
       auto& packet = transformed_packets.emplace_back(packets_arr[i]);
@@ -905,7 +1022,7 @@ void Queue::WriteInterceptor(const void* packets, uint64_t pkt_count, uint64_t u
         AddVendorSpecificPacket(profile.first->read_packet, &transformed_packets, interrupt_signal);
 
         // Added Interrupt Signal with barrier and provided handler for it
-        CreateBarrierPacket(interrupt_signal, &transformed_packets);
+        CreateBarrierPacket( &transformed_packets, &interrupt_signal, nullptr);
       } else {
         hsa_barrier_and_packet_t barrier{};
         barrier.header = HSA_PACKET_TYPE_BARRIER_AND << HSA_PACKET_HEADER_TYPE;
@@ -919,11 +1036,12 @@ void Queue::WriteInterceptor(const void* packets, uint64_t pkt_count, uint64_t u
               queue_info.GetGPUAgent().handle);
       //  Creating Async Handler to be called every time the interrupt signal is
       //  marked complete
-      SignalAsyncHandler(interrupt_signal,
-                         new queue_info_session_t{
-                             queue_info.GetGPUAgent(), session_id_snapshot, queue_info.GetQueueID(),
-                             writer_id, interrupt_signal, agentInfo.GetDeviceInfo().getGPUId(),
-                             agentInfo.GetDeviceInfo().getXccCount()});
+      SignalAsyncHandler(
+          interrupt_signal,
+          new queue_info_session_t{
+              queue_info.GetGPUAgent(), session_id_snapshot, queue_info.GetQueueID(), writer_id,
+              interrupt_signal, agentInfo.GetDeviceInfo().getGPUId(),
+              agentInfo.GetDeviceInfo().getXccCount(), queue_info.GetBlockSignal()});
       ACTIVE_INTERRUPT_SIGNAL_COUNT.fetch_add(1, std::memory_order_relaxed);
       // profile_id++;
       // } while (replay_mode_count > 0 && profile_id < replay_mode_count);  // Profiles loop end
@@ -989,7 +1107,7 @@ void Queue::WriteInterceptor(const void* packets, uint64_t pkt_count, uint64_t u
         dummy_signal.handle = 0;
         start_packet.header = HSA_PACKET_TYPE_VENDOR_SPECIFIC << HSA_PACKET_HEADER_TYPE;
         AddVendorSpecificPacket(&start_packet, &transformed_packets, dummy_signal);
-        CreateBarrierPacket(start_packet.completion_signal, &transformed_packets);
+        CreateBarrierPacket(&transformed_packets, &start_packet.completion_signal, nullptr) ;
       }
 
       auto& packet = transformed_packets.emplace_back(packets_arr[i]);
@@ -1035,7 +1153,7 @@ void Queue::WriteInterceptor(const void* packets, uint64_t pkt_count, uint64_t u
         AddVendorSpecificPacket(&stop_packet, &transformed_packets, interrupt_signal);
 
         // Added Interrupt Signal with barrier and provided handler for it
-        CreateBarrierPacket(interrupt_signal, &transformed_packets);
+        CreateBarrierPacket(&transformed_packets, &interrupt_signal, nullptr);
       } else {
         hsa_barrier_and_packet_t barrier{};
         barrier.header = HSA_PACKET_TYPE_BARRIER_AND << HSA_PACKET_HEADER_TYPE;
@@ -1064,11 +1182,40 @@ void Queue::WriteInterceptor(const void* packets, uint64_t pkt_count, uint64_t u
 
 
 Queue::Queue(const hsa_agent_t cpu_agent, const hsa_agent_t gpu_agent, hsa_queue_t* queue)
-    : cpu_agent_(cpu_agent), gpu_agent_(gpu_agent), intercept_queue_(queue) {}
+    : cpu_agent_(cpu_agent), gpu_agent_(gpu_agent), intercept_queue_(queue) {
+  state = is_destroy::normal;
+  CreateSignal(0, &block_signal_);
+  CreateSignal(0, &ready_signal_);
+  SignalAsyncReadyHandler(ready_signal_, this);
+}
 
 Queue::~Queue() {
-  while (ACTIVE_INTERRUPT_SIGNAL_COUNT.load(std::memory_order_acquire) > 0) {
+  std::unique_lock<std::mutex> queue_lock(qw_mutex);
+  {
+    profiler_serializer_t& serializer =
+        rocprofiler::ROCProfiler_Singleton::GetInstance().GetSerializer();
+    // std::cout << GetROCMToolObj()->
+    std::lock_guard<std::mutex> serializer_lock(serializer.serializer_mutex);
+    for (auto it = serializer.dispatch_ready.begin(); it != serializer.dispatch_ready.end();) {
+      if ((*it)->GetQueueID() == GetQueueID()) {
+        /*Deletes the queue to be destructed from the dispatch ready.*/
+        serializer.dispatch_ready.erase(it);
+        if (serializer.dispatch_queue->GetQueueID() == GetQueueID())
+          // ToDO [srnagara]: Need to find a solution rather than abort.
+          fatal("Queue is being destroyed while kernel launch is still active");
+      }
+    }
+    state = is_destroy::to_destroy;
+
+    rocprofiler::HSASupport_Singleton::GetInstance()
+        .GetCoreApiTable()
+        .hsa_signal_store_screlease_fn(ready_signal_, 0);
   }
+  this->cv_ready_signal.wait(queue_lock, [this] { return state == is_destroy::done_destroy; });
+
+  if (block_signal_.handle)
+    rocprofiler::HSASupport_Singleton::GetInstance().GetCoreApiTable().hsa_signal_destroy_fn(
+        block_signal_);
 }
 
 hsa_queue_t* Queue::GetCurrentInterceptQueue() { return intercept_queue_; }
@@ -1079,7 +1226,16 @@ hsa_agent_t Queue::GetCPUAgent() { return cpu_agent_; }
 
 uint64_t Queue::GetQueueID() { return intercept_queue_->id; }
 
-void CheckPacketReqiurements() { Packet::CheckPacketReqiurements(); }
+void CheckPacketReqiurements() {
+  Packet::CheckPacketReqiurements();
+}
+hsa_signal_t Queue::GetReadySignal() { return ready_signal_; }
+
+hsa_signal_t Queue::GetBlockSignal() { return block_signal_; }
+
+
+
+
 
 }  // namespace queue
 }  // namespace rocprofiler

projects/rocprofiler/src/core/hsa/queues/queue.h

@@ -34,7 +34,7 @@
 #include <mutex>
 #include <string>
 #include <vector>
-
+#include <condition_variable>
 #include "src/core/session/profiler/profiler.h"
 
 namespace rocprofiler {
@@ -49,6 +49,19 @@ uint32_t GetCurrentActiveInterruptSignalsCount();
 
 namespace queue {
 
+/* The enum here represents the
+state of the queue destruction.
+1. normal-The queue  destructor is not initiated.
+2. to_destroy - The queue destructor has been initiated.
+3. done_destroy - The async handler has been unregistered
+and the destructor can now complete.
+*/
+enum is_destroy {
+ normal=0,
+ to_destroy=1,
+ done_destroy=2
+};
+
 class Queue {
  public:
   Queue(const hsa_agent_t cpu_agent, const hsa_agent_t gpu_agent,
@@ -63,13 +76,20 @@ class Queue {
   uint64_t GetQueueID();
   static void PrintCounters();
   std::mutex qw_mutex;
+  enum is_destroy state;
+  std::condition_variable  cv_ready_signal;
+  hsa_signal_t GetReadySignal();
+  hsa_signal_t GetBlockSignal();
 
  private:
-  std::mutex mutex_;
+
   hsa_agent_t cpu_agent_;
   hsa_agent_t gpu_agent_;
   hsa_queue_t* intercept_queue_;
+  hsa_signal_t block_signal_;
+  hsa_signal_t ready_signal_;
 
+  bool unreg_async_handler_{false};
   hsa_status_t pmcCallback(hsa_ven_amd_aqlprofile_info_type_t info_type,
                            hsa_ven_amd_aqlprofile_info_data_t* info_data, void* data);
 };
@@ -82,6 +102,7 @@ struct queue_info_session_t {
   hsa_signal_t interrupt_signal;
   uint64_t gpu_index;
   uint32_t xcc_count;
+  hsa_signal_t block_signal;
 };
 
 void AddRecordCounters(rocprofiler_record_profiler_t* record, const pending_signal_t& pending);

projects/rocprofiler/src/core/session/session.cpp

@@ -46,9 +46,7 @@ Session::Session(rocprofiler_replay_mode_t replay_mode, rocprofiler_session_id_t
 }
 
 Session::~Session() {
-  while (GetCurrentActiveInterruptSignalsCount() > 0) {
-  }
-  {
+ {
     std::lock_guard<std::mutex> lock(session_lock_);
     if (FindFilterWithKind(ROCPROFILER_SPM_COLLECTION) && spmcounter_ &&
         spm_started_.load(std::memory_order_acquire)) {
@@ -204,8 +202,6 @@ void Session::Start() {
 
 void Session::Terminate() {
   if (is_active_) {
-    while (GetCurrentActiveInterruptSignalsCount() > 0) {
-    }
     rocprofiler::queue::ResetSessionID();
     std::lock_guard<std::mutex> lock(session_lock_);
     if (FindFilterWithKind(ROCPROFILER_SPM_COLLECTION)) {

projects/rocprofiler/tests-v2/featuretests/profiler/apps/hip_kernels.cpp

@@ -1,4 +1,5 @@
 #include <hip/hip_runtime.h>
+#include <vector>
 #ifdef NDEBUG
 #define HIP_ASSERT(x) x
 #else
@@ -14,6 +15,7 @@
 #define THREADS_PER_BLOCK_Y 16
 #define THREADS_PER_BLOCK_Z 1
 
+__device__ int counter = 0;
 // empty kernel
 __global__ void kernel() {}
 
@@ -31,13 +33,72 @@ __global__ void vectoradd_float(float* __restrict__ a, const float* __restrict__
   }
 }
 
+__global__ void add(int n, float* x, float* y) {
+
+  if(__hip_atomic_load(&counter, __ATOMIC_ACQUIRE, __HIP_MEMORY_SCOPE_AGENT) != 0){
+    abort();
+  }
+  __hip_atomic_fetch_add(&counter, 1, __ATOMIC_RELEASE, __HIP_MEMORY_SCOPE_SYSTEM);
+  int index = blockIdx.x * blockDim.x + threadIdx.x;
+  int stride = blockDim.x * gridDim.x;
+  for (int i = index; i < n; i += stride) y[i] = x[i] + y[i];
+   __hip_atomic_fetch_add(&counter, -1, __ATOMIC_RELEASE, __HIP_MEMORY_SCOPE_SYSTEM);
+
+}
+
 // launches an empty kernel in profiler context
 void KernelLaunch() {
   // run empty kernel
   kernel<<<1, 1>>>();
   hipDeviceSynchronize();
 }
+void LaunchMultiStreamKernels() {
+  int N = 1 << 4;
+  float* x = new float[N];
+  float* y = new float[N];
+  float* d_x;
+  float* d_y;
+  //   Allocate Unified Memory -- accessible from CPU or GPU
+  HIP_ASSERT(hipMallocManaged(&d_x, N * sizeof(float)));
+  HIP_ASSERT(hipMallocManaged(&d_y, N * sizeof(float)));
 
+  //   initialize x and y arrays on the host
+  for (int i = 0; i < N; i++) {
+    x[i] = 1.0f;
+    y[i] = 2.0f;
+  }
+  std::vector< hipStream_t> hip_streams;
+  for(int i = 0; i < 100; i++) {
+    hipStream_t stream;
+    hipStreamCreate	(&stream);
+    hip_streams.push_back(stream);
+
+  }
+  HIP_ASSERT(hipMemcpy(d_x, x, N * sizeof(float), hipMemcpyHostToDevice));
+  HIP_ASSERT(hipMemcpy(d_y, y, N * sizeof(float), hipMemcpyHostToDevice));
+
+  // Launch kernel on 1M elements on the GPU
+  int blockSize = 64;
+  // This Kernel will always be launched with one wave
+  int numBlocks = 1;
+  for(int i = 0; i < 100; i++) {
+    for(int j = 0; j < hip_streams.size(); j++)
+       hipLaunchKernelGGL(add, numBlocks, blockSize, 0, hip_streams[j], N, d_x, d_y);
+  }
+
+  //Wait for GPU to finish before accessing on host
+  HIP_ASSERT(hipDeviceSynchronize());
+
+  HIP_ASSERT(hipMemcpy(x, d_x, N * sizeof(float), hipMemcpyDeviceToHost));
+  HIP_ASSERT(hipMemcpy(y, d_y, N * sizeof(float), hipMemcpyDeviceToHost));
+
+    //   Free memory
+  HIP_ASSERT(hipFree(d_x));
+  HIP_ASSERT(hipFree(d_y));
+
+  delete[] x;
+  delete[] y;
+}
 int LaunchVectorAddKernel() {
   float* hostA;
   float* hostB;

projects/rocprofiler/tests-v2/featuretests/profiler/apps/hip_kernels.h

@@ -23,4 +23,5 @@ THE SOFTWARE.
 void vectoradd_float(float* a, const float* b, const float* c, int width, int height);
 void kernel();
 int LaunchVectorAddKernel();
-void KernelLaunch();
+void KernelLaunch();
+void LaunchMultiStreamKernels();

projects/rocprofiler/tests-v2/featuretests/profiler/profiler_gtest.cpp

@@ -887,6 +887,14 @@ class ProfilerAPITest : public ::testing::Test {
         const char* kernel_name_c = static_cast<const char*>(malloc(name_length * sizeof(char)));
         CheckApi(rocprofiler_query_kernel_info(ROCPROFILER_KERNEL_NAME, profiler_record->kernel_id,
                                                &kernel_name_c));
+        if (profiler_record->counters) {
+         for (uint64_t i = 0; i < profiler_record->counters_count.value; i++) {
+          if (profiler_record->counters[i].counter_handler.handle > 0) {
+           if(profiler_record->counters[i].value.value == 0)
+             rocprofiler::fatal("Serialization failed");
+        }
+      }
+    }
         // int gpu_index = profiler_record->gpu_id.handle;
         // uint64_t begin_time = profiler_record->timestamps.begin.value;
         // uint64_t end_time = profiler_record->timestamps.end.value;
@@ -958,6 +966,56 @@ TEST_F(ProfilerAPITest, WhenRunningMultipleThreadsProfilerAPIsWorkFine) {
   CheckApi(rocprofiler_finalize());
 }
 
+TEST_F(ProfilerAPITest, WhenRunningMultipleStreamsSerializationWorksFine) {
+  // set global path
+  init_test_path();
+
+  // Get the system cores
+  int num_cpu_cores = GetNumberOfCores();
+
+  // create as many threads as number of cores in system
+  std::vector<std::thread> threads(num_cpu_cores);
+
+  // initialize profiler by creating rocprofiler object
+  CheckApi(rocprofiler_initialize());
+
+  // Counter Collection with timestamps
+  rocprofiler_session_id_t session_id;
+  std::vector<const char*> counters;
+  counters.emplace_back("SQ_WAVES");
+
+  CheckApi(rocprofiler_create_session(ROCPROFILER_NONE_REPLAY_MODE, &session_id));
+
+  rocprofiler_buffer_id_t buffer_id;
+  CheckApi(rocprofiler_create_buffer(session_id, FlushCallback, 0x9999, &buffer_id));
+
+  rocprofiler_filter_id_t filter_id;
+  rocprofiler_filter_property_t property = {};
+  CheckApi(rocprofiler_create_filter(session_id, ROCPROFILER_COUNTERS_COLLECTION,
+                                     rocprofiler_filter_data_t{.counters_names = &counters[0]},
+                                     counters.size(), &filter_id, property));
+
+  CheckApi(rocprofiler_set_filter_buffer(session_id, filter_id, buffer_id));
+
+  // activating profiler session
+  CheckApi(rocprofiler_start_session(session_id));
+
+  LaunchMultiStreamKernels();
+  // deactivate session
+  CheckApi(rocprofiler_terminate_session(session_id));
+
+  // dump profiler data
+  CheckApi(rocprofiler_flush_data(session_id, buffer_id));
+
+  // destroy session
+  CheckApi(rocprofiler_destroy_session(session_id));
+
+  // finalize profiler by destroying rocprofiler object
+  CheckApi(rocprofiler_finalize());
+}
+
+
+
 /*
  * ###################################################
  * ############ Derived metrics tests ################