concurrent sqtt support

Change-Id: I91391fafabc93aefa5d244d870ef82b96a59dc52

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

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");

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 << ")");