concurrent sqtt support

Change-Id: I91391fafabc93aefa5d244d870ef82b96a59dc52


[ROCm/rocprofiler commit: 9f7e936d70]

projects/rocprofiler/inc/rocprofiler.h

@@ -72,6 +72,7 @@ typedef struct {
   uint64_t timeout;
   uint32_t timestamp_on;
   uint32_t hsa_intercepting;
+  uint32_t k_concurrent;
 } rocprofiler_settings_t;
 
 ////////////////////////////////////////////////////////////////////////////////

projects/rocprofiler/src/core/intercept_queue.cpp

@@ -42,4 +42,6 @@ InterceptQueue::queue_id_t InterceptQueue::current_queue_id = 0;
 rocprofiler_hsa_callback_fun_t InterceptQueue::submit_callback_fun_ = NULL;
 void* InterceptQueue::submit_callback_arg_ = NULL;
 
+bool InterceptQueue::k_concurrent_ = false;
+
 }  // namespace rocprofiler

projects/rocprofiler/src/core/intercept_queue.h

@@ -44,6 +44,17 @@ namespace rocprofiler {
 extern decltype(hsa_queue_create)* hsa_queue_create_fn;
 extern decltype(hsa_queue_destroy)* hsa_queue_destroy_fn;
 
+static std::mutex ctx_a_mutex;
+typedef std::map<Context*, bool> ctx_a_map_t;
+static ctx_a_map_t* ctx_a_map = NULL;
+static bool ck_ctx_inactive(Context* context) {
+  std::lock_guard<std::mutex> lock(ctx_a_mutex);
+  if (ctx_a_map == NULL) ctx_a_map = new ctx_a_map_t;
+  auto ret = ctx_a_map->insert({context, true});
+  if (ret.second == false) ctx_a_map->erase(context);
+  return ret.second;
+}
+
 class InterceptQueue {
  public:
   typedef std::recursive_mutex mutex_t;
@@ -79,7 +90,11 @@ class InterceptQueue {
     if (!obj_map_) obj_map_ = new obj_map_t;
     InterceptQueue* obj = new InterceptQueue(agent, *queue, proxy);
     (*obj_map_)[(uint64_t)(*queue)] = obj;
-    status = proxy->SetInterceptCB(OnSubmitCB, obj);
+    if (k_concurrent_) {
+      status = proxy->SetInterceptCB(OnSubmitCB_SQTT, obj);
+    } else {
+      status = proxy->SetInterceptCB(OnSubmitCB, obj);
+    }
     obj->queue_event_callback_ = callback;
     obj->queue_id = current_queue_id;
     ++current_queue_id;
@@ -251,6 +266,111 @@ class InterceptQueue {
     }
   }
 
+  static void OnSubmitCB_SQTT(const void* in_packets, uint64_t count, uint64_t user_que_idx, void* data,
+                         hsa_amd_queue_intercept_packet_writer writer) {
+    const packet_t* packets_arr = reinterpret_cast<const packet_t*>(in_packets);
+    InterceptQueue* obj = reinterpret_cast<InterceptQueue*>(data);
+    Queue* proxy = obj->proxy_;
+
+    if (submit_callback_fun_) {
+      mutex_.lock();
+      auto* callback_fun = submit_callback_fun_;
+      void* callback_arg = submit_callback_arg_;
+      mutex_.unlock();
+
+      if (callback_fun) {
+        for (uint64_t j = 0; j < count; ++j) {
+          const packet_t* packet = &packets_arr[j];
+          const hsa_kernel_dispatch_packet_t* dispatch_packet =
+              reinterpret_cast<const hsa_kernel_dispatch_packet_t*>(packet);
+
+          const char* kernel_name = NULL;
+          if (GetHeaderType(packet) == HSA_PACKET_TYPE_KERNEL_DISPATCH) {
+            uint64_t kernel_object = dispatch_packet->kernel_object;
+            const amd_kernel_code_t* kernel_code = GetKernelCode(kernel_object);
+            kernel_name = (GetHeaderType(packet) == HSA_PACKET_TYPE_KERNEL_DISPATCH) ?
+              QueryKernelName(kernel_object, kernel_code) : NULL;
+          }
+
+          // Prepareing submit callback data
+          rocprofiler_hsa_callback_data_t data{};
+          data.submit.packet = (void*)packet;
+          data.submit.kernel_name = kernel_name;
+          data.submit.queue = obj->queue_;
+          data.submit.device_type = obj->agent_info_->dev_type;
+          data.submit.device_id = obj->agent_info_->dev_index;
+
+          callback_fun(ROCPROFILER_HSA_CB_ID_SUBMIT, &data, callback_arg);
+        }
+      }
+    }
+
+    // Travers input packets
+    for (uint64_t j = 0; j < count; ++j) {
+      const packet_t* packet = &packets_arr[j];
+      bool to_submit = true;
+
+      // Checking for dispatch packet type
+      if ((GetHeaderType(packet) == HSA_PACKET_TYPE_KERNEL_DISPATCH) &&
+          (dispatch_callback_.load(std::memory_order_acquire) != NULL)) {
+        const hsa_kernel_dispatch_packet_t* dispatch_packet =
+            reinterpret_cast<const hsa_kernel_dispatch_packet_t*>(packet);
+        const hsa_signal_t completion_signal = dispatch_packet->completion_signal;
+
+        // Prepareing dispatch callback data
+        uint64_t kernel_object = dispatch_packet->kernel_object;
+        const amd_kernel_code_t* kernel_code = GetKernelCode(kernel_object);
+        const char* kernel_name = QueryKernelName(kernel_object, kernel_code);
+
+        rocprofiler_callback_data_t data = {obj->agent_info_->dev_id,
+                                            obj->agent_info_->dev_index,
+                                            obj->queue_,
+                                            user_que_idx,
+                                            obj->queue_id,
+                                            completion_signal,
+                                            dispatch_packet,
+                                            kernel_name,
+                                            kernel_object,
+                                            kernel_code,
+                                            (uint32_t)syscall(__NR_gettid),
+                                            NULL};
+
+        // Calling dispatch callback
+        rocprofiler_group_t group = {};
+        hsa_status_t status = (dispatch_callback_.load())(&data, callback_data_, &group);
+        free(const_cast<char*>(kernel_name));
+
+        // Injecting profiling start/stop packets
+        if ((status == HSA_STATUS_SUCCESS) && (group.context != NULL)) {
+          Context* context = reinterpret_cast<Context*>(group.context);
+          const bool ctx_inactive = ck_ctx_inactive(context);
+
+          const pkt_vector_t& start_vector = context->StartPackets(group.index);
+          const pkt_vector_t& stop_vector = context->StopPackets(group.index);
+          pkt_vector_t packets;
+          if (ctx_inactive) packets = start_vector;
+          packets.insert(packets.end(), *packet);
+          if (!ctx_inactive) packets.insert(packets.end(), stop_vector.begin(), stop_vector.end());
+          if (writer != NULL) {
+            writer(&packets[0], packets.size());
+          } else {
+            proxy->Submit(&packets[0], packets.size());
+          }
+          to_submit = false;
+        }
+      }
+
+      // Submitting the original packets if profiling was not enabled
+      if (to_submit) {
+        if (writer != NULL) {
+          writer(packet, 1);
+        } else {
+          proxy->Submit(packet, 1);
+        }
+      }
+    }
+  }
+
   static void SetCallbacks(rocprofiler_queue_callbacks_t callbacks, void* data) {
     std::lock_guard<mutex_t> lck(mutex_);
     if (callback_data_ != NULL) {
@@ -279,6 +399,8 @@ class InterceptQueue {
   static void TrackerOn(bool on) { tracker_on_ = on; }
   static bool IsTrackerOn() { return tracker_on_; }
 
+  static bool k_concurrent_;
+
  private:
   static void queue_event_callback(hsa_status_t status, hsa_queue_t *queue, void *arg) {
     if (status != HSA_STATUS_SUCCESS) EXC_ABORT(status, "queue error handling is not supported");

projects/rocprofiler/src/core/rocprofiler.cpp

@@ -216,6 +216,7 @@ uint32_t LoadTool() {
     if (settings.code_obj_tracking) intercept_mode |= CODE_OBJ_TRACKING_MODE;
     if (settings.memcopy_tracking) intercept_mode |= MEMCOPY_INTERCEPT_MODE;
     if (settings.hsa_intercepting) intercept_mode |= HSA_INTERCEPT_MODE;
+    if (settings.k_concurrent) InterceptQueue::k_concurrent_ = true;
   }
 
   ONLOAD_TRACE("end intercept_mode(" << intercept_mode << ")");

projects/rocprofiler/test/tool/tool.cpp

@@ -289,6 +289,20 @@ void dealloc_context_entry(context_entry_t* entry) {
   }
 }
 
+// Global context map
+static std::mutex ctx_a_mutex;
+typedef std::map<hsa_agent_handle_t, context_entry_t*> ctx_a_map_t;
+ctx_a_map_t* ctx_a_map = NULL;
+context_entry_t* ck_ctx_entry(hsa_agent_t agent, bool& found) {
+  std::lock_guard<std::mutex> lock(ctx_a_mutex);
+  if (ctx_a_map == NULL) ctx_a_map = new ctx_a_map_t;
+  auto ret = ctx_a_map->insert({agent.handle, NULL});
+  found = !ret.second;
+  if (found) ctx_a_map->erase(agent.handle);
+  else ret.first->second = new context_entry_t{};
+  return ret.first->second;
+}
+
 // Dump trace data to file
 void dump_sqtt_trace(const char* label, const uint32_t chunk, const void* data, const uint32_t& size) {
   if (result_prefix != NULL) {
@@ -481,34 +495,35 @@ bool dump_context_entry(context_entry_t* entry) {
   ++context_collected;
 
   const uint32_t index = entry->index;
-  FILE* file_handle = entry->file_handle;
-  const std::string nik_name = (to_truncate_names == 0) ? entry->data.kernel_name : filtr_kernel_name(entry->data.kernel_name);
-  const AgentInfo* agent_info = HsaRsrcFactory::Instance().GetAgentInfo(entry->agent);
-
-  fprintf(file_handle, "dispatch[%u], gpu-id(%u), queue-id(%u), queue-index(%lu), pid(%u), tid(%u), grd(%u), wgr(%u), lds(%u), scr(%u), vgpr(%u), sgpr(%u), fbar(%u), sig(0x%lx), kernel-name(\"%s\")",
-    index,
-    agent_info->dev_index,
-    entry->data.queue_id,
-    entry->data.queue_index,
-    my_pid,
-    entry->data.thread_id,
-    entry->kernel_properties.grid_size,
-    entry->kernel_properties.workgroup_size,
-    (entry->kernel_properties.lds_size + (AgentInfo::lds_block_size - 1)) & ~(AgentInfo::lds_block_size - 1),
-    entry->kernel_properties.scratch_size,
-    (entry->kernel_properties.vgpr_count + 1) * agent_info->vgpr_block_size,
-    (entry->kernel_properties.sgpr_count + agent_info->sgpr_block_dflt) * agent_info->sgpr_block_size,
-    entry->kernel_properties.fbarrier_count,
-    entry->kernel_properties.signal.handle,
-    nik_name.c_str());
-  if (record) fprintf(file_handle, ", time(%lu,%lu,%lu,%lu)",
-    record->dispatch,
-    record->begin,
-    record->end,
-    record->complete);
-  fprintf(file_handle, "\n");
-  fflush(file_handle);
-
+  if (index != UINT32_MAX) {
+    FILE* file_handle = entry->file_handle;
+    const std::string nik_name = (to_truncate_names == 0) ? entry->data.kernel_name : filtr_kernel_name(entry->data.kernel_name);
+    const AgentInfo* agent_info = HsaRsrcFactory::Instance().GetAgentInfo(entry->agent);
+  
+    fprintf(file_handle, "dispatch[%u], gpu-id(%u), queue-id(%u), queue-index(%lu), pid(%u), tid(%u), grd(%u), wgr(%u), lds(%u), scr(%u), vgpr(%u), sgpr(%u), fbar(%u), sig(0x%lx), kernel-name(\"%s\")",
+      index,
+      agent_info->dev_index,
+      entry->data.queue_id,
+      entry->data.queue_index,
+      my_pid,
+      entry->data.thread_id,
+      entry->kernel_properties.grid_size,
+      entry->kernel_properties.workgroup_size,
+      (entry->kernel_properties.lds_size + (AgentInfo::lds_block_size - 1)) & ~(AgentInfo::lds_block_size - 1),
+      entry->kernel_properties.scratch_size,
+      (entry->kernel_properties.vgpr_count + 1) * agent_info->vgpr_block_size,
+      (entry->kernel_properties.sgpr_count + agent_info->sgpr_block_dflt) * agent_info->sgpr_block_size,
+      entry->kernel_properties.fbarrier_count,
+      entry->kernel_properties.signal.handle,
+      nik_name.c_str());
+    if (record) fprintf(file_handle, ", time(%lu,%lu,%lu,%lu)",
+      record->dispatch,
+      record->begin,
+      record->end,
+      record->complete);
+    fprintf(file_handle, "\n");
+    fflush(file_handle);
+  }
   if (record) {
     delete record;
     entry->data.record = NULL;
@@ -606,6 +621,37 @@ bool context_handler(rocprofiler_group_t group, void* arg) {
   return false;
 }
 
+// Profiling completion handler for concurrent implementation
+// Dump the context entry
+// Return true if the context was dumped successfully
+bool context_handler_con(rocprofiler_group_t group, void* arg) {
+  context_entry_t* entry = reinterpret_cast<context_entry_t*>(arg);
+
+  if (pthread_mutex_lock(&mutex) != 0) {
+    perror("pthread_mutex_lock");
+    abort();
+  }
+
+  bool ret = true;
+  ret = dump_context_entry(entry);
+  if (ret == false) {
+    fprintf(stderr, "tool error: context is not complete\n");
+    abort();
+  }
+
+  if (trace_on) {
+    fprintf(stdout, "tool::handler_con: context_map %d tid %u\n", (int)(ctx_a_map->size()), GetTid());
+    fflush(stdout);
+  }
+
+  if (pthread_mutex_unlock(&mutex) != 0) {
+    perror("pthread_mutex_unlock");
+    abort();
+  }
+
+  return false;
+}
+
 bool check_filter(const rocprofiler_callback_data_t* callback_data, const callbacks_data_t* tool_data) {
   bool found = true;
 
@@ -734,6 +780,75 @@ hsa_status_t dispatch_callback(const rocprofiler_callback_data_t* callback_data,
   return status;
 }
 
+hsa_status_t dispatch_callback_con(const rocprofiler_callback_data_t* callback_data, void* user_data,
+                               rocprofiler_group_t* group) {
+  // Passed tool data
+  callbacks_data_t* tool_data = reinterpret_cast<callbacks_data_t*>(user_data);
+  // HSA status
+  hsa_status_t status = HSA_STATUS_ERROR;
+
+  // Checking dispatch condition
+  bool enabled = false;
+  if (tool_data->filter_on == 1) {
+    enabled = check_filter(callback_data, tool_data);
+    if (enabled == false) next_context_count();
+  }
+
+  // Checking context entry
+  bool found = false;
+  context_entry_t* entry = ck_ctx_entry(callback_data->agent, found);
+  if ((enabled == true) && (found == true)) return HSA_STATUS_SUCCESS;
+
+  if (found == false) {
+    *group = entry->group;
+  } else {
+    // Profiling context
+    rocprofiler_t* context = NULL;
+
+    // context properties
+    rocprofiler_properties_t properties{};
+    properties.handler = (result_prefix != NULL) ? context_handler_con : NULL;
+    properties.handler_arg = (void*)entry;
+  
+    rocprofiler_feature_t* features = tool_data->features;
+    unsigned feature_count = tool_data->feature_count;
+  
+    // Open profiling context
+    status = rocprofiler_open(callback_data->agent, features, feature_count,
+                              &context, 0 /*ROCPROFILER_MODE_SINGLEGROUP*/, &properties);
+    check_status(status);
+  
+    // Check that we have only one profiling group
+    uint32_t group_count = 0;
+    status = rocprofiler_group_count(context, &group_count);
+    check_status(status);
+    assert(group_count == 1);
+    // Get group[0]
+    const uint32_t group_index = 0;
+    status = rocprofiler_get_group(context, group_index, group);
+    check_status(status);
+  
+    // Fill profiling context entry
+    entry->index = UINT32_MAX;
+    entry->agent = callback_data->agent;
+    entry->group = *group;
+    entry->features = features;
+    entry->feature_count = feature_count;
+    entry->data = *callback_data;
+    entry->data.kernel_name = strdup(callback_data->kernel_name);
+    entry->file_handle = tool_data->file_handle;
+    entry->active = true;
+    reinterpret_cast<std::atomic<bool>*>(&entry->valid)->store(true);
+  
+    if (trace_on) {
+      fprintf(stdout, "tool::dispatch_con: context_map %d tid %u\n", (int)(ctx_a_map->size()), GetTid());
+      fflush(stdout);
+    }
+  }
+
+  return status;
+}
+
 hsa_status_t destroy_callback(hsa_queue_t* queue, void*) {
   results_output_break();
   dump_context_array(queue);
@@ -979,6 +1094,8 @@ extern "C" PUBLIC_API void OnLoadToolProp(rocprofiler_settings_t* settings)
   // Set HSA intercepting
   check_env_var("ROCP_HSA_INTERC", settings->hsa_intercepting);
   if (settings->hsa_intercepting) rocprofiler_set_hsa_callbacks(hsa_callbacks, (void*)14);
+  // Enable concurrent SQTT
+  check_env_var("ROCP_K_CONCURRENT", settings->k_concurrent);
 
   is_trace_local = settings->trace_local;
 
@@ -1118,6 +1235,8 @@ extern "C" PUBLIC_API void OnLoadToolProp(rocprofiler_settings_t* settings)
         HSA_VEN_AMD_AQLPROFILE_PARAMETER_NAME_SE_MASK;
     parameters_dict["SAMPLE_RATE"] =
         HSA_VEN_AMD_AQLPROFILE_PARAMETER_NAME_SAMPLE_RATE;
+    //parameters_dict["K_CON"] =
+    //    HSA_VEN_AMD_AQLPROFILE_PARAMETER_NAME_K_CONCURRENT;
 
     printf("    %s (", name.c_str());
     features[index] = {};
@@ -1133,6 +1252,8 @@ extern "C" PUBLIC_API void OnLoadToolProp(rocprofiler_settings_t* settings)
 
       if (tag != "parameters") fatal("ROCProfiler: trace node is not supported '" + tag + "'");
 
+      if (settings->k_concurrent != 0) parameter_count += 1;
+
       if (parameter_count != 0) {
         rocprofiler_parameter_t* parameters = new rocprofiler_parameter_t[parameter_count];
         unsigned p_index = 0;
@@ -1149,6 +1270,12 @@ extern "C" PUBLIC_API void OnLoadToolProp(rocprofiler_settings_t* settings)
           ++p_index;
         }
 
+        if (settings->k_concurrent != 0) {
+          parameters[parameter_count - 1] = {};
+          parameters[parameter_count - 1].parameter_name = HSA_VEN_AMD_AQLPROFILE_PARAMETER_NAME_K_CONCURRENT;
+          parameters[parameter_count - 1].value = 1;
+        }
+
         features[index].parameters = parameters;
         features[index].parameter_count = parameter_count;
       }
@@ -1173,7 +1300,11 @@ extern "C" PUBLIC_API void OnLoadToolProp(rocprofiler_settings_t* settings)
 
   // Adding dispatch observer
   rocprofiler_queue_callbacks_t callbacks_ptrs{0};
-  callbacks_ptrs.dispatch = dispatch_callback;
+  if (settings->k_concurrent != 0) {
+    callbacks_ptrs.dispatch = dispatch_callback_con;
+  } else {
+    callbacks_ptrs.dispatch = dispatch_callback;
+  }
   callbacks_ptrs.destroy = destroy_callback;
 
   callbacks_data = new callbacks_data_t{};