From c75faaaa3712599cf4aeb59151c4999fac7ae9aa Mon Sep 17 00:00:00 2001 From: Evgeny Date: Wed, 10 Jun 2020 11:00:18 -0500 Subject: [PATCH] concurrent sqtt support Change-Id: I91391fafabc93aefa5d244d870ef82b96a59dc52 [ROCm/rocprofiler commit: 9f7e936d7095c8b6c3377d0685be34955b5d098b] --- projects/rocprofiler/inc/rocprofiler.h | 1 + .../rocprofiler/src/core/intercept_queue.cpp | 2 + .../rocprofiler/src/core/intercept_queue.h | 124 +++++++++++- projects/rocprofiler/src/core/rocprofiler.cpp | 1 + projects/rocprofiler/test/tool/tool.cpp | 189 +++++++++++++++--- 5 files changed, 287 insertions(+), 30 deletions(-) diff --git a/projects/rocprofiler/inc/rocprofiler.h b/projects/rocprofiler/inc/rocprofiler.h index 24925cae85..8ed532ff6e 100644 --- a/projects/rocprofiler/inc/rocprofiler.h +++ b/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; //////////////////////////////////////////////////////////////////////////////// diff --git a/projects/rocprofiler/src/core/intercept_queue.cpp b/projects/rocprofiler/src/core/intercept_queue.cpp index 0b309d63ae..b6eabc6f5b 100644 --- a/projects/rocprofiler/src/core/intercept_queue.cpp +++ b/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 diff --git a/projects/rocprofiler/src/core/intercept_queue.h b/projects/rocprofiler/src/core/intercept_queue.h index a52d8c1d6b..93174017fe 100644 --- a/projects/rocprofiler/src/core/intercept_queue.h +++ b/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 ctx_a_map_t; +static ctx_a_map_t* ctx_a_map = NULL; +static bool ck_ctx_inactive(Context* context) { + std::lock_guard 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(in_packets); + InterceptQueue* obj = reinterpret_cast(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(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(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(kernel_name)); + + // Injecting profiling start/stop packets + if ((status == HSA_STATUS_SUCCESS) && (group.context != NULL)) { + Context* context = reinterpret_cast(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 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"); diff --git a/projects/rocprofiler/src/core/rocprofiler.cpp b/projects/rocprofiler/src/core/rocprofiler.cpp index 618edf2397..64493739f2 100644 --- a/projects/rocprofiler/src/core/rocprofiler.cpp +++ b/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 << ")"); diff --git a/projects/rocprofiler/test/tool/tool.cpp b/projects/rocprofiler/test/tool/tool.cpp index 1c526fbc0b..d8d43f0fca 100644 --- a/projects/rocprofiler/test/tool/tool.cpp +++ b/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 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 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(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(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*>(&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{};