From 1de44447f4e70d9532d0bea5dc7f7c0cd1593444 Mon Sep 17 00:00:00 2001 From: Benjamin Welton Date: Fri, 8 Mar 2024 07:02:43 -0800 Subject: [PATCH] Deadlock Fix for HSA and Serialization Disable/Enabling support (#582) * Initial barrier * Working on profiler serializer extraction * Current progress * Serializtion Support * source formatting (clang-format v11) (#583) Co-authored-by: bwelton <1683479+bwelton@users.noreply.github.com> * cmake formatting (cmake-format) (#584) Co-authored-by: bwelton <1683479+bwelton@users.noreply.github.com> * Minor fix * Current Progress * Current progress * More fixes * Serialization Fixes * Bug fix * source formatting (clang-format v11) (#600) Co-authored-by: bwelton <1683479+bwelton@users.noreply.github.com> * More fixes * More minor fixes * source formatting (clang-format v11) (#603) Co-authored-by: bwelton <1683479+bwelton@users.noreply.github.com> * source formatting (clang-format v11) (#604) Co-authored-by: bwelton <1683479+bwelton@users.noreply.github.com> * Lock order inversion false positive * order fix * More changes * source formatting (clang-format v11) (#607) Co-authored-by: bwelton <1683479+bwelton@users.noreply.github.com> * minor test fix * Minor test changes --------- Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> Co-authored-by: bwelton <1683479+bwelton@users.noreply.github.com> --- source/lib/common/synchronized.hpp | 1 - source/lib/rocprofiler-sdk/counters/core.cpp | 116 ++++-- source/lib/rocprofiler-sdk/counters/core.hpp | 12 +- .../rocprofiler-sdk/counters/tests/core.cpp | 25 +- source/lib/rocprofiler-sdk/hsa/CMakeLists.txt | 14 +- .../lib/rocprofiler-sdk/hsa/hsa_barrier.cpp | 111 ++++++ .../lib/rocprofiler-sdk/hsa/hsa_barrier.hpp | 84 ++++ .../hsa/profile_serializer.cpp | 247 ++++++++++++ .../hsa/profile_serializer.hpp | 98 +++++ source/lib/rocprofiler-sdk/hsa/queue.cpp | 29 +- source/lib/rocprofiler-sdk/hsa/queue.hpp | 19 +- .../rocprofiler-sdk/hsa/queue_controller.cpp | 158 ++------ .../rocprofiler-sdk/hsa/queue_controller.hpp | 56 ++- .../lib/rocprofiler-sdk/tests/CMakeLists.txt | 2 +- .../lib/rocprofiler-sdk/tests/hsa_barrier.cpp | 358 ++++++++++++++++++ source/scripts/thread-sanitizer-suppr.txt | 3 + 16 files changed, 1099 insertions(+), 234 deletions(-) create mode 100644 source/lib/rocprofiler-sdk/hsa/hsa_barrier.cpp create mode 100644 source/lib/rocprofiler-sdk/hsa/hsa_barrier.hpp create mode 100644 source/lib/rocprofiler-sdk/hsa/profile_serializer.cpp create mode 100644 source/lib/rocprofiler-sdk/hsa/profile_serializer.hpp create mode 100644 source/lib/rocprofiler-sdk/tests/hsa_barrier.cpp diff --git a/source/lib/common/synchronized.hpp b/source/lib/common/synchronized.hpp index 5f0341d39c..37ab31f304 100644 --- a/source/lib/common/synchronized.hpp +++ b/source/lib/common/synchronized.hpp @@ -72,7 +72,6 @@ public: // Do not allow this data structure to be copied, std::move only. Synchronized(const Synchronized&) = delete; - Synchronized& operator=(const Synchronized&) = delete; template decltype(auto) rlock(FuncT&& lambda, Args&&... args) const; diff --git a/source/lib/rocprofiler-sdk/counters/core.cpp b/source/lib/rocprofiler-sdk/counters/core.cpp index c7a7f0b2c5..ba4c5c9870 100644 --- a/source/lib/rocprofiler-sdk/counters/core.cpp +++ b/source/lib/rocprofiler-sdk/counters/core.cpp @@ -253,7 +253,8 @@ counter_callback_info::get_packet(std::unique_ptr& * We return an AQLPacket containing the start/stop/read packets for injection. */ std::unique_ptr -queue_cb(const std::shared_ptr& info, +queue_cb(const context::context* ctx, + const std::shared_ptr& info, const hsa::Queue& queue, const hsa::rocprofiler_packet& pkt, uint64_t kernel_id, @@ -261,7 +262,43 @@ queue_cb(const std::shared_ptr& info, const hsa::Queue::queue_info_session_t::external_corr_id_map_t& extern_corr_ids, const context::correlation_id* correlation_id) { - if(!info || !info->user_cb) return nullptr; + CHECK(info && ctx); + + // Maybe adds serialization packets to the AQLPacket (if serializer is enabled) + // and maybe adds barrier packets if the state is transitioning from serialized <-> + // unserialized + auto maybe_add_serialization = [&](auto& gen_pkt) { + hsa::get_queue_controller().serializer().rlock([&](const auto& serializer) { + for(auto& s_pkt : serializer.kernel_dispatch(queue)) + { + gen_pkt->before_krn_pkt.push_back(s_pkt.ext_amd_aql_pm4); + } + }); + }; + + // Packet generated when no instrumentation is performed. May contain serialization + // packets/barrier packets (and can be empty). + auto no_instrumentation = [&]() { + auto ret_pkt = std::make_unique(nullptr); + // If we have a counter collection context but it is not enabled, we still might need + // to add barrier packets to transition from serialized -> unserialized execution. This + // transition is coordinated by the serializer. + maybe_add_serialization(ret_pkt); + info->packet_return_map.wlock([&](auto& data) { data.emplace(ret_pkt.get(), nullptr); }); + return ret_pkt; + }; + + if(!ctx || !ctx->counter_collection) return nullptr; + + bool is_enabled = false; + + ctx->counter_collection->enabled.rlock( + [&](const auto& collect_ctx) { is_enabled = collect_ctx; }); + + if(!is_enabled || !info->user_cb) + { + return no_instrumentation(); + } auto _corr_id_v = rocprofiler_correlation_id_t{.internal = 0, .external = context::null_user_data}; @@ -294,7 +331,10 @@ queue_cb(const std::shared_ptr& info, info->user_cb(dispatch_data, &req_profile, user_data, info->callback_args); - if(req_profile.handle == 0) return nullptr; + if(req_profile.handle == 0) + { + return no_instrumentation(); + } auto prof_config = get_controller().get_profile_cfg(req_profile); CHECK(prof_config); @@ -303,27 +343,13 @@ queue_cb(const std::shared_ptr& info, auto status = info->get_packet(ret_pkt, queue.get_agent(), prof_config); CHECK_EQ(status, ROCPROFILER_STATUS_SUCCESS) << rocprofiler_get_status_string(status); - if(ret_pkt->empty) return ret_pkt; - - auto&& CreateBarrierPacket = - [](hsa_signal_t* dependency_signal, - hsa_signal_t* completion_signal, - common::container::small_vector& _packets) { - hsa::rocprofiler_packet barrier{}; - barrier.barrier_and.header = HSA_PACKET_TYPE_BARRIER_AND << HSA_PACKET_HEADER_TYPE; - if(dependency_signal != nullptr) barrier.barrier_and.dep_signal[0] = *dependency_signal; - if(completion_signal != nullptr) - barrier.barrier_and.completion_signal = *completion_signal; - _packets.emplace_back(barrier.ext_amd_aql_pm4); - }; - - hsa_signal_t ready_signal = queue.ready_signal; - hsa_signal_t block_signal = queue.block_signal; - CreateBarrierPacket(nullptr, &ready_signal, ret_pkt->before_krn_pkt); - CreateBarrierPacket(&block_signal, &block_signal, ret_pkt->before_krn_pkt); + maybe_add_serialization(ret_pkt); + if(ret_pkt->empty) + { + return ret_pkt; + } ret_pkt->before_krn_pkt.push_back(ret_pkt->start); - ret_pkt->before_krn_pkt.end()->completion_signal.handle = 0; ret_pkt->after_krn_pkt.push_back(ret_pkt->stop); ret_pkt->after_krn_pkt.push_back(ret_pkt->read); for(auto& aql_pkt : ret_pkt->after_krn_pkt) @@ -338,13 +364,14 @@ queue_cb(const std::shared_ptr& info, * Callback called by HSA interceptor when the kernel has completed processing. */ void -completed_cb(const std::shared_ptr& info, +completed_cb(const context::context* ctx, + const std::shared_ptr& info, const hsa::Queue& queue, hsa::rocprofiler_packet, const hsa::Queue::queue_info_session_t& session, inst_pkt_t& pkts) { - if(!info || pkts.empty()) return; + CHECK(info && ctx); std::shared_ptr prof_config; // Get the Profile Config @@ -365,10 +392,11 @@ completed_cb(const std::shared_ptr& info, if(!pkt) return; - if(!pkt->empty) - { - hsa::profiler_serializer_kernel_completion_signal(session.queue.block_signal); - } + hsa::get_queue_controller().serializer().wlock( + [&](auto& serializer) { serializer.kernel_completion_signal(session.queue); }); + + // We have no profile config, nothing to output. + if(!prof_config) return; auto decoded_pkt = EvaluateAST::read_pkt(prof_config->pkt_generator.get(), *pkt); EvaluateAST::read_special_counters( @@ -452,13 +480,21 @@ start_context(const context::context* ctx) auto& controller = hsa::get_queue_controller(); - // Only one thread should be attempting to enable/disable this context + bool already_enabled = true; + controller.enable_serialization(); ctx->counter_collection->enabled.wlock([&](auto& enabled) { if(enabled) return; + already_enabled = false; + enabled = true; + }); + + if(!already_enabled) + { for(auto& cb : ctx->counter_collection->callbacks) { // Insert our callbacks into HSA Interceptor. This // turns on counter instrumentation. + if(cb->queue_id != rocprofiler::hsa::ClientID{-1}) continue; cb->queue_id = controller.add_callback( std::nullopt, [=](const hsa::Queue& q, @@ -467,17 +503,22 @@ start_context(const context::context* ctx) rocprofiler_user_data_t* user_data, const hsa::Queue::queue_info_session_t::external_corr_id_map_t& extern_corr_ids, const context::correlation_id* correlation_id) { - return queue_cb( - cb, q, kern_pkt, kernel_id, user_data, extern_corr_ids, correlation_id); + return queue_cb(ctx, + cb, + q, + kern_pkt, + kernel_id, + user_data, + extern_corr_ids, + correlation_id); }, // Completion CB [=](const hsa::Queue& q, hsa::rocprofiler_packet kern_pkt, const hsa::Queue::queue_info_session_t& session, - inst_pkt_t& aql) { completed_cb(cb, q, kern_pkt, session, aql); }); + inst_pkt_t& aql) { completed_cb(ctx, cb, q, kern_pkt, session, aql); }); } - enabled = true; - }); + } } void @@ -489,14 +530,9 @@ stop_context(const context::context* ctx) ctx->counter_collection->enabled.wlock([&](auto& enabled) { if(!enabled) return; - for(auto& cb : ctx->counter_collection->callbacks) - { - // Remove our callbacks from HSA's queue controller - controller.remove_callback(cb->queue_id); - cb->queue_id = -1; - } enabled = false; }); + controller.disable_serialization(); } bool diff --git a/source/lib/rocprofiler-sdk/counters/core.hpp b/source/lib/rocprofiler-sdk/counters/core.hpp index 8fe76c47b3..4b288f145e 100644 --- a/source/lib/rocprofiler-sdk/counters/core.hpp +++ b/source/lib/rocprofiler-sdk/counters/core.hpp @@ -126,7 +126,8 @@ void stop_context(const context::context*); std::unique_ptr -queue_cb(const std::shared_ptr& info, +queue_cb(const context::context* ctx, + const std::shared_ptr& info, const hsa::Queue& queue, const hsa::rocprofiler_packet& pkt, uint64_t kernel_id, @@ -139,11 +140,12 @@ using inst_pkt_t = common::container:: small_vector, ClientID>, 4>; void -completed_cb(const std::shared_ptr&, - const hsa::Queue&, +completed_cb(const context::context* ctx, + const std::shared_ptr& info, + const hsa::Queue& queue, hsa::rocprofiler_packet, - const hsa::Queue::queue_info_session_t&, - inst_pkt_t& pkts); + const hsa::Queue::queue_info_session_t& session, + inst_pkt_t& pkts); std::shared_ptr get_profile_config(rocprofiler_profile_config_id_t); } // namespace counters diff --git a/source/lib/rocprofiler-sdk/counters/tests/core.cpp b/source/lib/rocprofiler-sdk/counters/tests/core.cpp index 10cf96b03f..a5243b46a9 100644 --- a/source/lib/rocprofiler-sdk/counters/tests/core.cpp +++ b/source/lib/rocprofiler-sdk/counters/tests/core.cpp @@ -414,6 +414,10 @@ TEST(core, check_callbacks) context::push_client(1); ROCPROFILER_CALL(rocprofiler_create_context(&get_client_ctx()), "context creation failed"); + context::context ctx; + ctx.counter_collection = std::make_unique(); + ctx.counter_collection->enabled.wlock([](auto& data) { data = true; }); + auto agents = hsa::get_queue_controller().get_supported_agents(); ASSERT_GT(agents.size(), 0); hsa::get_queue_controller().disable_serialization(); @@ -471,9 +475,10 @@ TEST(core, check_callbacks) expected.agent_id = fq.get_agent().get_rocp_agent()->id; hsa::Queue::queue_info_session_t::external_corr_id_map_t extern_ids = {}; + auto user_data = rocprofiler_user_data_t{.value = corr_id.internal}; auto ret_pkt = counters::queue_cb( - cb_info, fq, pkt, expected.kernel_id, &user_data, extern_ids, &corr_id); + &ctx, cb_info, fq, pkt, expected.kernel_id, &user_data, extern_ids, &corr_id); ASSERT_TRUE(ret_pkt) << fmt::format("Expected a packet to be generated for - {}", metric.name()); @@ -515,7 +520,7 @@ TEST(core, check_callbacks) counters::inst_pkt_t pkts; pkts.emplace_back( std::make_pair(std::move(ret_pkt), static_cast(0))); - completed_cb(cb_info, fq, pkt, sess, pkts); + completed_cb(&ctx, cb_info, fq, pkt, sess, pkts); rocprofiler_flush_buffer(opt_buff_id); rocprofiler_destroy_buffer(opt_buff_id); } @@ -621,14 +626,6 @@ TEST(core, start_stop_buffered_ctx) */ ROCPROFILER_CALL(rocprofiler_stop_context(get_client_ctx()), "stop context"); - found = false; - hsa::get_queue_controller().iterate_callbacks([&](auto cid, const auto&) { - if(cid == ctx.counter_collection->callbacks.at(0)->queue_id) - { - found = true; - } - }); - EXPECT_FALSE(found); found = false; ctx.counter_collection->enabled.rlock([&](const auto& data) { found = data; }); EXPECT_FALSE(found); @@ -694,14 +691,6 @@ TEST(core, start_stop_callback_ctx) */ ROCPROFILER_CALL(rocprofiler_stop_context(get_client_ctx()), "stop context"); - found = false; - hsa::get_queue_controller().iterate_callbacks([&](auto cid, const auto&) { - if(cid == ctx.counter_collection->callbacks.at(0)->queue_id) - { - found = true; - } - }); - EXPECT_FALSE(found); found = false; ctx.counter_collection->enabled.rlock([&](const auto& data) { found = data; }); EXPECT_FALSE(found); diff --git a/source/lib/rocprofiler-sdk/hsa/CMakeLists.txt b/source/lib/rocprofiler-sdk/hsa/CMakeLists.txt index 5e0e562260..0d2e91802f 100644 --- a/source/lib/rocprofiler-sdk/hsa/CMakeLists.txt +++ b/source/lib/rocprofiler-sdk/hsa/CMakeLists.txt @@ -1,6 +1,14 @@ # -set(ROCPROFILER_LIB_HSA_SOURCES agent_cache.cpp aql_packet.cpp async_copy.cpp - code_object.cpp hsa.cpp queue_controller.cpp queue.cpp) +set(ROCPROFILER_LIB_HSA_SOURCES + agent_cache.cpp + aql_packet.cpp + async_copy.cpp + code_object.cpp + hsa.cpp + queue_controller.cpp + queue.cpp + hsa_barrier.cpp + profile_serializer.cpp) set(ROCPROFILER_LIB_HSA_HEADERS agent_cache.hpp @@ -8,7 +16,9 @@ set(ROCPROFILER_LIB_HSA_HEADERS async_copy.hpp code_object.hpp defines.hpp + hsa_barrier.hpp hsa.hpp + profile_serializer.hpp queue_controller.hpp queue.hpp types.hpp diff --git a/source/lib/rocprofiler-sdk/hsa/hsa_barrier.cpp b/source/lib/rocprofiler-sdk/hsa/hsa_barrier.cpp new file mode 100644 index 0000000000..c3a214d79d --- /dev/null +++ b/source/lib/rocprofiler-sdk/hsa/hsa_barrier.cpp @@ -0,0 +1,111 @@ +// 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 "lib/rocprofiler-sdk/hsa/hsa_barrier.hpp" + +namespace rocprofiler +{ +namespace hsa +{ +void +hsa_barrier::set_barrier(queue_map_t& q) +{ + _core_api.hsa_signal_store_screlease_fn(_barrier_signal, 1); + _queue_waiting.wlock([&](auto& queue_waiting) { + for(auto& [_, queue] : q) + { + queue->lock_queue([ptr = queue.get(), &queue_waiting]() { + if(ptr->active_async_packets() > 0) + { + queue_waiting[ptr->get_id().handle] = ptr->active_async_packets(); + } + }); + } + if(queue_waiting.empty()) + { + _barried_finished(); + _core_api.hsa_signal_store_screlease_fn(_barrier_signal, 0); + } + }); +} + +std::optional +hsa_barrier::enqueue_packet(const Queue* queue) +{ + if(_complete) return std::nullopt; + bool return_block = false; + _barrier_enqueued.wlock([&](auto& barrier_enqueued) { + if(barrier_enqueued.find(queue->get_id().handle) == barrier_enqueued.end()) + { + return_block = true; + barrier_enqueued.insert(queue->get_id().handle); + } + }); + + if(!return_block) return std::nullopt; + + rocprofiler_packet barrier{}; + barrier.barrier_and.header = HSA_PACKET_TYPE_BARRIER_AND << HSA_PACKET_HEADER_TYPE; + barrier.barrier_and.dep_signal[0] = _barrier_signal; + LOG(ERROR) << "Barrier Added: " << _barrier_signal.handle; + return barrier; +} + +void +hsa_barrier::remove_queue(const Queue* queue) +{ + _queue_waiting.wlock([&](auto& queue_waiting) { + if(queue_waiting.find(queue->get_id().handle) == queue_waiting.end()) return; + queue_waiting.erase(queue->get_id().handle); + if(queue_waiting.empty()) + { + _barried_finished(); + _complete = true; + _core_api.hsa_signal_store_screlease_fn(_barrier_signal, 0); + } + }); +} + +bool +hsa_barrier::register_completion(const Queue* queue) +{ + bool found = false; + _queue_waiting.wlock([&](auto& queue_waiting) { + if(queue_waiting.find(queue->get_id().handle) == queue_waiting.end()) return; + found = true; + queue_waiting[queue->get_id().handle]--; + if(queue_waiting[queue->get_id().handle] == 0) + { + queue_waiting.erase(queue->get_id().handle); + if(queue_waiting.empty()) + { + _barried_finished(); + // We are done, release the barrier + _complete = true; + _core_api.hsa_signal_store_screlease_fn(_barrier_signal, 0); + } + } + }); + return found; +} +} // namespace hsa +} // namespace rocprofiler diff --git a/source/lib/rocprofiler-sdk/hsa/hsa_barrier.hpp b/source/lib/rocprofiler-sdk/hsa/hsa_barrier.hpp new file mode 100644 index 0000000000..543f4c7ad0 --- /dev/null +++ b/source/lib/rocprofiler-sdk/hsa/hsa_barrier.hpp @@ -0,0 +1,84 @@ +// 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. + +#pragma once + +#include +#include +#include +#include + +#include +#include +#include + +#include + +#include "lib/common/synchronized.hpp" +#include "lib/rocprofiler-sdk/hsa/queue.hpp" + +namespace rocprofiler +{ +namespace hsa +{ +class hsa_barrier +{ +public: + using queue_map_t = std::unordered_map>; + hsa_barrier(std::function&& finished, CoreApiTable core_api) + : _barried_finished(std::move(finished)) + , _core_api(core_api) + { + // Create the barrier signal + _core_api.hsa_signal_create_fn(0, 0, nullptr, &_barrier_signal); + } + + ~hsa_barrier() + { + // Destroy the barrier signal + _core_api.hsa_signal_store_screlease_fn(_barrier_signal, 0); + _core_api.hsa_signal_destroy_fn(_barrier_signal); + } + + void set_barrier(queue_map_t& q); + + std::optional enqueue_packet(const Queue* queue); + bool register_completion(const Queue* queue); + + bool complete() const { return _core_api.hsa_signal_load_scacquire_fn(_barrier_signal) == 0; } + + // Removes a queue from the barrier dependency list + void remove_queue(const Queue* queue); + +private: + std::function _barried_finished; + CoreApiTable _core_api; + common::Synchronized> _queue_waiting; + common::Synchronized> _barrier_enqueued; + std::atomic _complete{false}; + + // Blocks all queues from executing until the barrier is lifted + hsa_signal_t _barrier_signal{}; +}; + +} // namespace hsa +} // namespace rocprofiler diff --git a/source/lib/rocprofiler-sdk/hsa/profile_serializer.cpp b/source/lib/rocprofiler-sdk/hsa/profile_serializer.cpp new file mode 100644 index 0000000000..91acf343d3 --- /dev/null +++ b/source/lib/rocprofiler-sdk/hsa/profile_serializer.cpp @@ -0,0 +1,247 @@ +// 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 "lib/rocprofiler-sdk/hsa/profile_serializer.hpp" + +#include "lib/rocprofiler-sdk/hsa/queue_controller.hpp" + +namespace rocprofiler +{ +namespace hsa +{ +namespace +{ +bool +profiler_serializer_ready_signal_handler(hsa_signal_value_t /* signal_value */, void* data) +{ + auto* hsa_queue = static_cast(data); + const auto* queue = get_queue_controller().get_queue(*hsa_queue); + CHECK(queue); + get_queue_controller().serializer().wlock( + [&](auto& serializer) { serializer.queue_ready(hsa_queue, *queue); }); + return true; +} + +void +clear_complete_barriers(std::deque& barriers) +{ + while(!barriers.empty()) + { + if(barriers.front().barrier->complete()) + { + barriers.pop_front(); + } + else + { + break; + } + } +} + +} // namespace + +void +profiler_serializer::add_queue(hsa_queue_t** hsa_queues, const Queue& queue) +{ + hsa_signal_t signal = queue.ready_signal; + hsa_status_t status = get_queue_controller().get_ext_table().hsa_amd_signal_async_handler_fn( + signal, HSA_SIGNAL_CONDITION_EQ, -1, profiler_serializer_ready_signal_handler, *hsa_queues); + if(status != HSA_STATUS_SUCCESS) LOG(FATAL) << "hsa_amd_signal_async_handler failed"; +} + +void +profiler_serializer::kernel_completion_signal(const Queue& completed) +{ + // We do not want to track kernel compleiton signals before we have reached the barrier + clear_complete_barriers(_barrier); + + // Find the state of this barrier + auto state = _serializer_status.load(); + bool found = false; + for(auto& barrier : _barrier) + { + // Register completion of the kernel. Each queue has a number of kernels it is + // waiting on to complete for each barrier. If more than one barrier is present + // that has this queue, then it will contain a count that is the sum of all previous + // kernel packets in the queue. Thus we must register completion with every barrier. + // The state of the queue at this time is the state of the first barrier (or the state + // of the serializer if no barriers are present). + if(barrier.barrier->register_completion(&completed) && !found) + { + state = barrier.state; + found = true; + } + } + + if(state == Status::DISABLED) return; + + CHECK(_dispatch_queue); + _dispatch_queue = nullptr; + get_queue_controller().get_core_table().hsa_signal_store_screlease_fn(completed.block_signal, + 1); + get_queue_controller().get_core_table().hsa_signal_store_screlease_fn(completed.ready_signal, + 0); + if(!_dispatch_ready.empty()) + { + const auto* queue = _dispatch_ready.front(); + _dispatch_ready.erase(_dispatch_ready.begin()); + get_queue_controller().get_core_table().hsa_signal_store_screlease_fn(queue->block_signal, + 0); + _dispatch_queue = queue; + } +} + +void +profiler_serializer::queue_ready(hsa_queue_t* hsa_queue, const Queue& queue) +{ + { + std::lock_guard cv_lock(queue.cv_mutex); + if(queue.get_state() == queue_state::to_destroy) + { + get_queue_controller().set_queue_state(queue_state::done_destroy, hsa_queue); + get_queue_controller().get_core_table().hsa_signal_destroy_fn(queue.ready_signal); + queue.cv_ready_signal.notify_one(); + return; + } + } + get_queue_controller().get_core_table().hsa_signal_store_screlease_fn(queue.ready_signal, 1); + if(_dispatch_queue == nullptr) + { + get_queue_controller().get_core_table().hsa_signal_store_screlease_fn(queue.block_signal, + 0); + _dispatch_queue = &queue; + } + else + { + _dispatch_ready.push_back(&queue); + } +} + +common::container::small_vector +profiler_serializer::kernel_dispatch(const Queue& queue) const +{ + common::container::small_vector ret; + auto&& CreateBarrierPacket = [](hsa_signal_t* dependency_signal, + hsa_signal_t* completion_signal) { + hsa::rocprofiler_packet barrier{}; + barrier.barrier_and.header = HSA_PACKET_TYPE_BARRIER_AND << HSA_PACKET_HEADER_TYPE; + barrier.barrier_and.header |= HSA_FENCE_SCOPE_SYSTEM + << HSA_PACKET_HEADER_SCACQUIRE_FENCE_SCOPE; + barrier.barrier_and.header |= HSA_FENCE_SCOPE_SYSTEM + << HSA_PACKET_HEADER_SCRELEASE_FENCE_SCOPE; + barrier.barrier_and.header |= 1 << HSA_PACKET_HEADER_BARRIER; + if(dependency_signal != nullptr) barrier.barrier_and.dep_signal[0] = *dependency_signal; + if(completion_signal != nullptr) barrier.barrier_and.completion_signal = *completion_signal; + return barrier; + }; + + if(!_barrier.empty()) + { + if(auto maybe_barrier = _barrier.back().barrier->enqueue_packet(&queue)) + { + ret.push_back(*maybe_barrier); + } + } + + switch(_serializer_status) + { + case Status::DISABLED: return ret; + case Status::ENABLED: + { + hsa_signal_t ready_signal = queue.ready_signal; + hsa_signal_t block_signal = queue.block_signal; + ret.push_back(CreateBarrierPacket(&ready_signal, &ready_signal)); + ret.push_back(CreateBarrierPacket(&block_signal, &block_signal)); + break; + }; + } + return ret; +} + +void +profiler_serializer::destory_queue(hsa_queue_t* id, const Queue& queue) +{ + /*Deletes the queue to be destructed from the dispatch ready.*/ + for(auto& barriers : _barrier) + { + barriers.barrier->remove_queue(&queue); + } + + _dispatch_ready.erase( + std::remove_if( + _dispatch_ready.begin(), + _dispatch_ready.end(), + [&](auto& it) { + /*Deletes the queue to be destructed from the dispatch ready.*/ + if(it->get_id().handle == queue.get_id().handle) + { + if(_dispatch_queue && _dispatch_queue->get_id().handle == queue.get_id().handle) + { + // insert fatal condition here + // ToDO [srnagara]: Need to find a solution rather than abort. + LOG(FATAL) + << "Queue is being destroyed while kernel launch is still active"; + } + return true; + } + return false; + }), + _dispatch_ready.end()); + get_queue_controller().set_queue_state(queue_state::to_destroy, id); + get_queue_controller().get_core_table().hsa_signal_store_screlease_fn(queue.ready_signal, 0); +} + +// Enable the serializer +void +profiler_serializer::enable(queue_map_t& queues) +{ + if(_serializer_status == Status::ENABLED) return; + _serializer_status = Status::ENABLED; + if(queues.empty()) return; + + clear_complete_barriers(_barrier); + + _barrier.emplace_back( + Status::DISABLED, + std::make_unique([] {}, get_queue_controller().get_core_table())); + _serializer_status = Status::ENABLED; + _barrier.back().barrier->set_barrier(queues); +} + +// Disable the serializer +void +profiler_serializer::disable(queue_map_t& queues) +{ + if(_serializer_status == Status::DISABLED) return; + _serializer_status = Status::DISABLED; + if(queues.empty()) return; + clear_complete_barriers(_barrier); + + _barrier.emplace_back( + Status::ENABLED, + std::make_unique([] {}, get_queue_controller().get_core_table())); + _serializer_status = Status::DISABLED; + _barrier.back().barrier->set_barrier(queues); +} + +} // namespace hsa +} // namespace rocprofiler \ No newline at end of file diff --git a/source/lib/rocprofiler-sdk/hsa/profile_serializer.hpp b/source/lib/rocprofiler-sdk/hsa/profile_serializer.hpp new file mode 100644 index 0000000000..5a71b540db --- /dev/null +++ b/source/lib/rocprofiler-sdk/hsa/profile_serializer.hpp @@ -0,0 +1,98 @@ +// 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. + +#pragma once + +#include + +#include "lib/common/container/small_vector.hpp" +#include "lib/rocprofiler-sdk/hsa/hsa_barrier.hpp" +#include "lib/rocprofiler-sdk/hsa/queue.hpp" + +#include +#include +#include +#include +#include + +namespace rocprofiler +{ +namespace hsa +{ +/*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. +*/ +class profiler_serializer +{ +public: + enum class Status + { + ENABLED, + DISABLED, + }; + + struct barrier_with_state + { + barrier_with_state(Status _state, std::unique_ptr _barrier) + : state(_state) + , barrier(std::move(_barrier)) + {} + Status state; + std::unique_ptr barrier; + }; + + using queue_map_t = std::unordered_map>; + void kernel_completion_signal(const Queue&); + // Signal a kernel dispatch is taking place, generates packets needed to be + // inserted to support kernel dispatch + common::container::small_vector kernel_dispatch(const Queue&) const; + + void queue_ready(hsa_queue_t* hsa_queue, const Queue& queue); + // Enable the serializer + void enable(queue_map_t& queues); + // Disable the serializer + void disable(queue_map_t& queues); + + void destory_queue(hsa_queue_t* id, const Queue& queue); + + static void add_queue(hsa_queue_t** hsa_queues, const Queue& queue); + +private: + const Queue* _dispatch_queue{nullptr}; + std::deque _dispatch_ready; + std::atomic _serializer_status{Status::DISABLED}; + std::deque _barrier; +}; + +} // namespace hsa +} // namespace rocprofiler diff --git a/source/lib/rocprofiler-sdk/hsa/queue.cpp b/source/lib/rocprofiler-sdk/hsa/queue.cpp index 468c359fc6..e140cdef80 100644 --- a/source/lib/rocprofiler-sdk/hsa/queue.cpp +++ b/source/lib/rocprofiler-sdk/hsa/queue.cpp @@ -26,6 +26,7 @@ #include "lib/rocprofiler-sdk/context/context.hpp" #include "lib/rocprofiler-sdk/hsa/code_object.hpp" #include "lib/rocprofiler-sdk/hsa/hsa.hpp" +#include "lib/rocprofiler-sdk/hsa/queue_controller.hpp" #include #include @@ -171,6 +172,12 @@ AsyncSignalHandler(hsa_signal_value_t /*signal_v*/, void* data) // Delete signals and packets, signal we have completed. if(queue_info_session.interrupt_signal.handle != 0u) { +#if !defined(NDEBUG) + hsa::get_queue_controller()._debug_signals.wlock( + [&](auto& signals) { signals.erase(queue_info_session.interrupt_signal.handle); }); +#endif + hsa::get_core_table()->hsa_signal_store_screlease_fn(queue_info_session.interrupt_signal, + -1); hsa::get_core_table()->hsa_signal_destroy_fn(queue_info_session.interrupt_signal); } if(queue_info_session.kernel_pkt.ext_amd_aql_pm4.completion_signal.handle != 0u) @@ -234,6 +241,7 @@ WriteInterceptor(const void* packets, std::vector& _packets) { hsa_barrier_and_packet_t barrier{}; barrier.header = HSA_PACKET_TYPE_BARRIER_AND << HSA_PACKET_HEADER_TYPE; + barrier.header |= 1 << HSA_PACKET_HEADER_BARRIER; if(dependency_signal != nullptr) barrier.dep_signal[0] = *dependency_signal; if(completion_signal != nullptr) barrier.completion_signal = *completion_signal; _packets.emplace_back(barrier); @@ -285,6 +293,7 @@ WriteInterceptor(const void* packets, continue; } + queue.async_started(); // Copy kernel pkt, copy is to allow for signal to be modified rocprofiler_packet kernel_pkt = packets_arr[i]; uint64_t kernel_id = get_kernel_id(kernel_pkt.kernel_dispatch.kernel_object); @@ -322,9 +331,7 @@ WriteInterceptor(const void* packets, // Barrier packet is last packet inserted into queue if(inserted_before) { - CreateBarrierPacket(&transformed_packets.back().ext_amd_aql_pm4.completion_signal, - nullptr, - transformed_packets); + CreateBarrierPacket(nullptr, nullptr, transformed_packets); } transformed_packets.emplace_back(kernel_pkt); @@ -334,7 +341,10 @@ WriteInterceptor(const void* packets, if(original_packet.completion_signal.handle != 0u) { hsa_barrier_and_packet_t barrier{}; - barrier.header = HSA_PACKET_TYPE_BARRIER_AND << HSA_PACKET_HEADER_TYPE; + barrier.header = HSA_PACKET_TYPE_BARRIER_AND << HSA_PACKET_HEADER_TYPE; + // barrier.header |= HSA_FENCE_SCOPE_SYSTEM << HSA_PACKET_HEADER_SCACQUIRE_FENCE_SCOPE; + // barrier.header |= HSA_FENCE_SCOPE_SYSTEM << HSA_PACKET_HEADER_SCRELEASE_FENCE_SCOPE; + // barrier.header |= 1 << HSA_PACKET_HEADER_BARRIER; barrier.completion_signal = original_packet.completion_signal; transformed_packets.emplace_back(barrier); } @@ -356,10 +366,11 @@ WriteInterceptor(const void* packets, if(injected_end_pkt) { transformed_packets.back().ext_amd_aql_pm4.completion_signal = interrupt_signal; - CreateBarrierPacket(&interrupt_signal, nullptr, transformed_packets); + CreateBarrierPacket(&interrupt_signal, &interrupt_signal, transformed_packets); } else { + get_core_table()->hsa_signal_store_screlease_fn(interrupt_signal, 0); hsa_barrier_and_packet_t barrier{}; barrier.header = HSA_PACKET_TYPE_BARRIER_AND << HSA_PACKET_HEADER_TYPE; barrier.completion_signal = interrupt_signal; @@ -373,7 +384,6 @@ WriteInterceptor(const void* packets, // Enqueue the signal into the handler. Will call completed_cb when // signal completes. - queue.async_started(); queue.signal_async_handler( interrupt_signal, new Queue::queue_info_session_t{.queue = queue, @@ -400,8 +410,12 @@ Queue::~Queue() { sync(); } void Queue::signal_async_handler(const hsa_signal_t& signal, Queue::queue_info_session_t* data) const { +#if !defined(NDEBUG) + hsa::get_queue_controller()._debug_signals.wlock( + [&](auto& signals) { signals[signal.handle] = signal; }); +#endif hsa_status_t status = _ext_api.hsa_amd_signal_async_handler_fn( - signal, HSA_SIGNAL_CONDITION_EQ, 0, AsyncSignalHandler, static_cast(data)); + signal, HSA_SIGNAL_CONDITION_EQ, -1, AsyncSignalHandler, static_cast(data)); LOG_IF(FATAL, status != HSA_STATUS_SUCCESS && status != HSA_STATUS_INFO_BREAK) << "Error: hsa_amd_signal_async_handler failed"; } @@ -451,6 +465,7 @@ Queue::Queue(const AgentCache& agent, create_signal(0, &ready_signal); create_signal(0, &block_signal); + _core_api.hsa_signal_store_screlease_fn(ready_signal, 0); *queue = _intercept_queue; } diff --git a/source/lib/rocprofiler-sdk/hsa/queue.hpp b/source/lib/rocprofiler-sdk/hsa/queue.hpp index c2bbee1f7b..78a1cb3a8a 100644 --- a/source/lib/rocprofiler-sdk/hsa/queue.hpp +++ b/source/lib/rocprofiler-sdk/hsa/queue.hpp @@ -173,6 +173,9 @@ public: template void signal_callback(FuncT&& func) const; + template + void lock_queue(FuncT&& func); + virtual rocprofiler_queue_id_t get_id() const; // Fast check to see if we have any callbacks we need to notify @@ -181,9 +184,10 @@ public: // Tracks the number of in flight kernel executions we // are waiting on. We cannot destroy Queue until all kernels // have comleted. - void async_started() { _active_async_packets++; } - void async_complete() { _active_async_packets--; } - void sync() const; + void async_started() { _active_async_packets++; } + void async_complete() { _active_async_packets--; } + int64_t active_async_packets() const { return _active_async_packets; } + void sync() const; void register_callback(ClientID id, queue_cb_t enqueue_cb, completed_cb_t complete_cb); void remove_callback(ClientID id); @@ -206,6 +210,7 @@ private: rocprofiler::common::Synchronized _callbacks = {}; hsa_queue_t* _intercept_queue = nullptr; queue_state _state = queue_state::normal; + std::mutex _lock_queue; }; inline rocprofiler_queue_id_t @@ -221,5 +226,13 @@ Queue::signal_callback(FuncT&& func) const _callbacks.rlock([&func](const auto& data) { func(data); }); } +template +void +Queue::lock_queue(FuncT&& func) +{ + std::unique_lock lock(_lock_queue); + func(); +} + } // namespace hsa } // namespace rocprofiler diff --git a/source/lib/rocprofiler-sdk/hsa/queue_controller.cpp b/source/lib/rocprofiler-sdk/hsa/queue_controller.cpp index 65f737b674..7c65d6e9bd 100644 --- a/source/lib/rocprofiler-sdk/hsa/queue_controller.cpp +++ b/source/lib/rocprofiler-sdk/hsa/queue_controller.cpp @@ -62,8 +62,8 @@ create_queue(hsa_agent_t agent, get_queue_controller().get_ext_table(), queue); - get_queue_controller().profiler_serializer_register_ready_signal_handler( - new_queue->ready_signal, *queue); + get_queue_controller().serializer().wlock( + [&](auto& serializer) { serializer.add_queue(queue, *new_queue); }); get_queue_controller().add_queue(*queue, std::move(new_queue)); return HSA_STATUS_SUCCESS; @@ -110,42 +110,9 @@ QueueController::destroy_queue(hsa_queue_t* id) { const auto* queue = get_queue_controller().get_queue(*id); std::unique_lock cvlock(queue->cv_mutex); - profiler_serializer([&](auto& data) { - /*Deletes the queue to be destructed from the dispatch ready.*/ - data.dispatch_ready.erase( - std::remove_if( - data.dispatch_ready.begin(), - data.dispatch_ready.end(), - [&](auto& it) { - /*Deletes the queue to be destructed from the dispatch ready.*/ - if(it->get_id().handle == queue->get_id().handle) - { - if(data.dispatch_queue && - data.dispatch_queue->get_id().handle == queue->get_id().handle) - { - // insert fatal condition here - // ToDO [srnagara]: Need to find a solution rather than abort. - LOG(FATAL) - << "Queue is being destroyed while kernel launch is still active"; - } - return true; - } - return false; - }), - data.dispatch_ready.end()); - set_queue_state(queue_state::to_destroy, id); - /* - This lambda triggers the async ready handler. - The async ready handler then unregisters itself - and sets the queue state to done_destroy for which - the condition variable here is waiting for. - */ - auto trigger_ready_async_handler = [queue]() { - get_queue_controller().get_core_table().hsa_signal_store_screlease_fn( - queue->ready_signal, 0); - }; - trigger_ready_async_handler(); - }); + + serializer().wlock([&](auto& serializer) { serializer.destory_queue(id, *queue); }); + queue->cv_ready_signal.wait( cvlock, [queue] { return queue->get_state() == queue_state::done_destroy; }); if(queue->block_signal.handle != 0) @@ -262,85 +229,44 @@ QueueController::get_queue(const hsa_queue_t& _hsa_queue) const _hsa_queue); } -template -void -QueueController::profiler_serializer(FuncT&& lambda) -{ - _profiler_serializer.wlock(std::forward(lambda)); -} - void QueueController::disable_serialization() { - profiler_serializer([](auto& serializer) { serializer.enabled = false; }); -} - -namespace -{ -/* - 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 -profiler_serializer_ready_signal_handler(hsa_signal_value_t /* signal_value */, void* data) -{ - auto* hsa_queue = static_cast(data); - const auto* queue = get_queue_controller().get_queue(*hsa_queue); - get_queue_controller().profiler_serializer([&](auto& serializer) { - if(!serializer.enabled) return; - { - std::lock_guard cv_lock(queue->cv_mutex); - if(queue->get_state() == queue_state::to_destroy) - { - get_queue_controller().set_queue_state(queue_state::done_destroy, hsa_queue); - get_queue_controller().get_core_table().hsa_signal_destroy_fn(queue->ready_signal); - queue->cv_ready_signal.notify_one(); - return; - } - } - get_queue_controller().get_core_table().hsa_signal_store_screlease_fn(queue->ready_signal, - 1); - if(serializer.dispatch_queue == nullptr) - { - get_queue_controller().get_core_table().hsa_signal_store_screlease_fn( - queue->block_signal, 0); - serializer.dispatch_queue = queue; - } - else - { - serializer.dispatch_ready.push_back(queue); - } + _queues.wlock([](queue_map_t& _queues_v) { + get_queue_controller().serializer().wlock( + [&](auto& serializer) { serializer.disable(_queues_v); }); }); - return true; } -} // namespace void -profiler_serializer_kernel_completion_signal(hsa_signal_t queue_block_signal) +QueueController::enable_serialization() { - get_queue_controller().profiler_serializer([queue_block_signal](auto& serializer) { - if(!serializer.enabled) return; - assert(serializer.dispatch_queue != nullptr); - serializer.dispatch_queue = nullptr; - get_queue_controller().get_core_table().hsa_signal_store_screlease_fn(queue_block_signal, - 1); - if(!serializer.dispatch_ready.empty()) + _queues.wlock([](queue_map_t& _queues_v) { + get_queue_controller().serializer().wlock( + [&](auto& serializer) { serializer.enable(_queues_v); }); + }); +} + +void +QueueController::print_debug_signals() const +{ +#if !defined(NDEBUG) + _debug_signals.rlock([&](const auto& signals) { + for(const auto& [id, signal] : signals) { - auto queue = serializer.dispatch_ready.front(); - serializer.dispatch_ready.erase(serializer.dispatch_ready.begin()); - get_queue_controller().get_core_table().hsa_signal_store_screlease_fn( - queue->block_signal, 0); - serializer.dispatch_queue = queue; + LOG(ERROR) << "Signal " << signal.handle << " " + << get_core_table().hsa_signal_load_scacquire_fn(signal); + } + }); +#endif + + _queues.rlock([&](const auto& queues) { + for(const auto& [_, queue] : queues) + { + LOG(ERROR) << "Queue " << queue->get_id().handle << " " << queue->ready_signal.handle + << ":" << get_core_table().hsa_signal_load_scacquire_fn(queue->ready_signal) + << " " << queue->block_signal.handle << ":" + << get_core_table().hsa_signal_load_scacquire_fn(queue->block_signal); } }); } @@ -351,22 +277,6 @@ QueueController::set_queue_state(enum queue_state state, hsa_queue_t* hsa_queue) _queues.wlock([&](auto& map) { map[hsa_queue]->set_state(state); }); } -/* - 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 -QueueController::profiler_serializer_register_ready_signal_handler(const hsa_signal_t& signal, - void* data) const -{ - hsa_status_t status = get_ext_table().hsa_amd_signal_async_handler_fn( - signal, HSA_SIGNAL_CONDITION_EQ, 0, profiler_serializer_ready_signal_handler, data); - if(status != HSA_STATUS_SUCCESS) LOG(FATAL) << "hsa_amd_signal_async_handler failed"; -} - void QueueController::iterate_queues(const queue_iterator_cb_t& cb) const { diff --git a/source/lib/rocprofiler-sdk/hsa/queue_controller.hpp b/source/lib/rocprofiler-sdk/hsa/queue_controller.hpp index a2bce3d45c..dcb853044d 100644 --- a/source/lib/rocprofiler-sdk/hsa/queue_controller.hpp +++ b/source/lib/rocprofiler-sdk/hsa/queue_controller.hpp @@ -24,6 +24,7 @@ #include +#include "lib/rocprofiler-sdk/hsa/profile_serializer.hpp" #include "lib/rocprofiler-sdk/hsa/queue.hpp" #include @@ -36,27 +37,6 @@ namespace rocprofiler { namespace hsa { -/*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 -{ - const Queue* dispatch_queue{nullptr}; - std::deque dispatch_ready; - bool enabled{true}; -}; - // Tracks and manages HSA queues class QueueController { @@ -65,6 +45,7 @@ public: std::tuple; using queue_iterator_cb_t = std::function; using callback_iterator_cb_t = std::function; + using queue_map_t = std::unordered_map>; QueueController() = default; // Initializes the QueueInterceptor. This must be delayed until @@ -86,7 +67,7 @@ public: const CoreApiTable& get_core_table() const { return _core_table; } const AmdExtTable& get_ext_table() const { return _ext_table; } - // Gets the list of supported HSA agents that can be intercepted + // Gets the list of supported HSA agents that can be Pintercepted const auto& get_supported_agents() const { return _supported_agents; } auto& get_supported_agents() { return _supported_agents; } @@ -94,32 +75,41 @@ public: void iterate_queues(const queue_iterator_cb_t&) const; void set_queue_state(queue_state state, hsa_queue_t* hsa_queue); - void profiler_serializer_register_ready_signal_handler(const hsa_signal_t& signal, - void* data) const; + void add_dispatch_ready(const Queue* queue); - template - void profiler_serializer(FuncT&& lambda); void iterate_callbacks(const callback_iterator_cb_t&) const; + common::Synchronized& serializer() { return _profiler_serializer; } + /** * Disable serialization for QueueController, has no effect if counter collection * is not in use (which defaults to no serialization mechanism). Should only be used for * testing. */ + void enable_serialization(); void disable_serialization(); + // Prints current state of signals for queues, used for debugging. Only prints + // serialization related signals if not compiled in debug mode. + void print_debug_signals() const; + +#if !defined(NDEBUG) + // Tracks the creation of all signals in queues, used for debugging and disabled + // in release mode (adds locking around signal creation). + common::Synchronized> _debug_signals; +#endif + private: - using queue_map_t = std::unordered_map>; using client_id_map_t = std::unordered_map; using agent_cache_map_t = std::unordered_map; - CoreApiTable _core_table = {}; - AmdExtTable _ext_table = {}; - common::Synchronized _queues = {}; - common::Synchronized _callback_cache = {}; - agent_cache_map_t _supported_agents = {}; - common::Synchronized _profiler_serializer; + CoreApiTable _core_table = {}; + AmdExtTable _ext_table = {}; + common::Synchronized _queues = {}; + common::Synchronized _callback_cache = {}; + agent_cache_map_t _supported_agents = {}; + common::Synchronized _profiler_serializer; }; QueueController& diff --git a/source/lib/rocprofiler-sdk/tests/CMakeLists.txt b/source/lib/rocprofiler-sdk/tests/CMakeLists.txt index c1b36813a3..e489a1c8e4 100644 --- a/source/lib/rocprofiler-sdk/tests/CMakeLists.txt +++ b/source/lib/rocprofiler-sdk/tests/CMakeLists.txt @@ -12,7 +12,7 @@ include(GoogleTest) # -------------------------------------------------------------------------------------- # set(rocprofiler_lib_sources agent.cpp buffer.cpp hsa.cpp naming.cpp timestamp.cpp - version.cpp) + version.cpp hsa_barrier.cpp) add_executable(rocprofiler-lib-tests) target_sources(rocprofiler-lib-tests PRIVATE ${rocprofiler_lib_sources} details/agent.cpp) diff --git a/source/lib/rocprofiler-sdk/tests/hsa_barrier.cpp b/source/lib/rocprofiler-sdk/tests/hsa_barrier.cpp new file mode 100644 index 0000000000..f5aaa77de2 --- /dev/null +++ b/source/lib/rocprofiler-sdk/tests/hsa_barrier.cpp @@ -0,0 +1,358 @@ +#include +#include + +#include +#include +#include +#include + +#include + +#include "lib/rocprofiler-sdk/agent.hpp" +#include "lib/rocprofiler-sdk/context/context.hpp" +#include "lib/rocprofiler-sdk/hsa/agent_cache.hpp" +#include "lib/rocprofiler-sdk/hsa/hsa_barrier.hpp" +#include "lib/rocprofiler-sdk/hsa/queue_controller.hpp" +#include "lib/rocprofiler-sdk/registration.hpp" +#include "rocprofiler-sdk/registration.h" + +using namespace rocprofiler; +using namespace rocprofiler::hsa; + +namespace rocprofiler +{ +namespace hsa +{ +class FakeQueue : public Queue +{ +public: + FakeQueue(const AgentCache& a, rocprofiler_queue_id_t id) + : Queue(a) + , _agent(a) + , _id(id) + {} + virtual const AgentCache& get_agent() const override final { return _agent; }; + virtual rocprofiler_queue_id_t get_id() const override final { return _id; }; + + ~FakeQueue() {} + +private: + const AgentCache& _agent; + rocprofiler_queue_id_t _id = {}; +}; + +} // namespace hsa +} // namespace rocprofiler + +namespace +{ +AmdExtTable& +get_ext_table() +{ + static auto _v = []() { + auto val = AmdExtTable{}; + val.hsa_amd_memory_pool_get_info_fn = hsa_amd_memory_pool_get_info; + val.hsa_amd_agent_iterate_memory_pools_fn = hsa_amd_agent_iterate_memory_pools; + val.hsa_amd_memory_pool_allocate_fn = hsa_amd_memory_pool_allocate; + val.hsa_amd_memory_pool_free_fn = hsa_amd_memory_pool_free; + val.hsa_amd_agent_memory_pool_get_info_fn = hsa_amd_agent_memory_pool_get_info; + val.hsa_amd_agents_allow_access_fn = hsa_amd_agents_allow_access; + return val; + }(); + return _v; +} + +CoreApiTable& +get_api_table() +{ + static auto _v = []() { + auto val = CoreApiTable{}; + val.hsa_iterate_agents_fn = hsa_iterate_agents; + val.hsa_agent_get_info_fn = hsa_agent_get_info; + val.hsa_queue_create_fn = hsa_queue_create; + val.hsa_queue_destroy_fn = hsa_queue_destroy; + val.hsa_signal_create_fn = hsa_signal_create; + val.hsa_signal_destroy_fn = hsa_signal_destroy; + val.hsa_signal_store_screlease_fn = hsa_signal_store_screlease; + val.hsa_signal_load_scacquire_fn = hsa_signal_load_scacquire; + return val; + }(); + return _v; +} + +QueueController::queue_map_t +create_queue_map(size_t count) +{ + QueueController::queue_map_t ret; + + auto agents = hsa::get_queue_controller().get_supported_agents(); + + for(size_t i = 0; i < count; i++) + { + auto& agent_cache = agents.begin()->second; + // Create queue + hsa_queue_t* queue; + hsa_queue_create( + agent_cache.get_hsa_agent(), 2048, HSA_QUEUE_TYPE_SINGLE, NULL, NULL, 0, 0, &queue); + ret[queue] = std::make_unique(agent_cache, rocprofiler_queue_id_t{.handle = i}); + } + + return ret; +} + +std::atomic should_execute_handler{false}; +std::atomic executed_handlers{0}; +bool +barrier_signal_handler(hsa_signal_value_t, void* data) +{ + CHECK(data); + CHECK(should_execute_handler) << "Signal handler called when it should not have been"; + hsa_signal_destroy(*static_cast(data)); + delete static_cast(data); + executed_handlers++; + return false; +} + +// Injects a barrier packet into the queue followed by a packet with an async handler +// associated with it. If the barrier is not released, the async handler should not +// be executed (checked with should_execute_handler). +void +inject_barriers(hsa_barrier& barrier, QueueController::queue_map_t& queues) +{ + auto packet_store_release = [](uint32_t* packet, uint16_t header, uint16_t rest) { + __atomic_store_n(packet, header | (rest << 16), __ATOMIC_RELEASE); + }; + + auto header_pkt = [](hsa_packet_type_t type) { + uint16_t header = type << HSA_PACKET_HEADER_TYPE; + header |= HSA_FENCE_SCOPE_SYSTEM << HSA_PACKET_HEADER_SCACQUIRE_FENCE_SCOPE; + header |= HSA_FENCE_SCOPE_SYSTEM << HSA_PACKET_HEADER_SCRELEASE_FENCE_SCOPE; + return header; + }; + + auto enqueue_pkt = [&](auto& queue, auto& packets, auto& pkt) { + uint64_t packet_id = hsa_queue_add_write_index_screlease(queue, 1); + while(packet_id - hsa_queue_load_read_index_scacquire(queue) >= queue->size) + ; + hsa_barrier_and_packet_t* packet = packets + packet_id % queue->size; + (*packet) = pkt; + packet_store_release((uint32_t*) packet, header_pkt(HSA_PACKET_TYPE_BARRIER_AND), 0); + hsa_signal_store_screlease(queue->doorbell_signal, packet_id); + }; + + for(auto& [hsa_queue, fq] : queues) + { + auto pkt = barrier.enqueue_packet(fq.get()); + ASSERT_EQ(pkt.has_value(), true); + hsa_barrier_and_packet_t* packets = (hsa_barrier_and_packet_t*) hsa_queue->base_address; + enqueue_pkt(hsa_queue, packets, pkt->barrier_and); + + // Construct packet that will trigger async handler after barrier is released + rocprofiler_packet post_barrier{}; + hsa_signal_t* completion_signal = new hsa_signal_t; + hsa_signal_create(1, 0, nullptr, completion_signal); + post_barrier.barrier_and.header = HSA_PACKET_TYPE_BARRIER_AND << HSA_PACKET_HEADER_TYPE; + post_barrier.barrier_and.completion_signal = *completion_signal; + hsa_amd_signal_async_handler(*completion_signal, + HSA_SIGNAL_CONDITION_EQ, + 0, + barrier_signal_handler, + static_cast(completion_signal)); + enqueue_pkt(hsa_queue, packets, post_barrier.barrier_and); + } + + // Ensure that the barrier packet is reached on all queues + usleep(100); +} + +void +test_init() +{ + HsaApiTable table; + table.amd_ext_ = &get_ext_table(); + table.core_ = &get_api_table(); + rocprofiler::agent::construct_agent_cache(&table); + hsa::get_queue_controller().init(get_api_table(), get_ext_table()); +} +} // namespace + +TEST(hsa_barrier, no_block_single) +{ + ASSERT_EQ(hsa_init(), HSA_STATUS_SUCCESS); + test_init(); + + registration::init_logging(); + registration::set_init_status(-1); + context::push_client(1); + + bool complete = false; + auto finished_func = [&]() { complete = true; }; + + auto queues = create_queue_map(1); + + // Immediate return of barrier due to no active async packets + rocprofiler::hsa::hsa_barrier barrier(finished_func, get_api_table()); + barrier.set_barrier(queues); + executed_handlers = 0; + ASSERT_TRUE(barrier.complete()); + should_execute_handler = true; + inject_barriers(barrier, queues); + ASSERT_EQ(complete, true); + while(executed_handlers != 1) + { + usleep(10); + } + + registration::set_init_status(1); + registration::finalize(); +} + +TEST(hsa_barrier, no_block_multi) +{ + ASSERT_EQ(hsa_init(), HSA_STATUS_SUCCESS); + test_init(); + + registration::init_logging(); + registration::set_init_status(-1); + context::push_client(1); + + bool complete = false; + auto finished_func = [&]() { complete = true; }; + + auto queues = create_queue_map(10); + + // Immediate return of barrier due to no active async packets + rocprofiler::hsa::hsa_barrier barrier(finished_func, get_api_table()); + barrier.set_barrier(queues); + ASSERT_TRUE(barrier.complete()); + should_execute_handler = true; + executed_handlers = 0; + inject_barriers(barrier, queues); + ASSERT_EQ(complete, true); + while(executed_handlers != 10) + { + usleep(10); + } + + registration::set_init_status(1); + registration::finalize(); +} + +TEST(hsa_barrier, block_single) +{ + std::vector pkt_waiting; + ASSERT_EQ(hsa_init(), HSA_STATUS_SUCCESS); + test_init(); + + registration::init_logging(); + registration::set_init_status(-1); + context::push_client(1); + + bool complete = false; + auto finished_func = [&]() { complete = true; }; + + auto queues = create_queue_map(1); + + rocprofiler::hsa::hsa_barrier barrier(finished_func, get_api_table()); + + // Simulate waiting on packets already in the queue to complete + for(auto& [_, queue] : queues) + { + pkt_waiting.push_back(queue.get()); + queue->async_started(); + } + should_execute_handler = false; + executed_handlers = 0; + + barrier.set_barrier(queues); + ASSERT_FALSE(barrier.complete()); + + should_execute_handler = false; + executed_handlers = 0; + inject_barriers(barrier, queues); + + ASSERT_EQ(complete, false); + should_execute_handler = true; + + for(auto& queue : pkt_waiting) + { + queue->async_complete(); + barrier.register_completion(queue); + } + + ASSERT_EQ(complete, true); + // Wait for the signal handlers to execute. If we deadlock here, + // we are not triggering the completion of the signal handler. + while(executed_handlers != 1) + { + usleep(100); + } + + registration::set_init_status(1); + registration::finalize(); +} + +TEST(hsa_barrier, block_multi) +{ + std::vector pkt_waiting; + ASSERT_EQ(hsa_init(), HSA_STATUS_SUCCESS); + test_init(); + + registration::init_logging(); + registration::set_init_status(-1); + context::push_client(1); + + bool complete = false; + auto finished_func = [&]() { complete = true; }; + + auto queues = create_queue_map(10); + + // Immediate return of barrier due to no active async packets + rocprofiler::hsa::hsa_barrier barrier(finished_func, get_api_table()); + + // Simulate waiting on packets already in the queue to complete + for(auto& [_, queue] : queues) + { + for(size_t i = 0; i < 30; i++) + { + pkt_waiting.push_back(queue.get()); + queue->async_started(); + } + } + should_execute_handler = false; + executed_handlers = 0; + + barrier.set_barrier(queues); + ASSERT_FALSE(barrier.complete()); + + should_execute_handler = false; + executed_handlers = 0; + inject_barriers(barrier, queues); + + ASSERT_EQ(complete, false); + + auto rng = std::default_random_engine{}; + std::shuffle(std::begin(pkt_waiting), std::end(pkt_waiting), rng); + for(size_t i = 0; i < pkt_waiting.size(); i++) + { + ASSERT_EQ(complete, false); + ASSERT_FALSE(barrier.complete()); + if(i == pkt_waiting.size() - 1) + { + should_execute_handler = true; + } + pkt_waiting[i]->async_complete(); + barrier.register_completion(pkt_waiting[i]); + } + + ASSERT_EQ(complete, true); + // Wait for the signal handlers to execute. If we deadlock here, + // we are not triggering the completion of the signal handler. + while(executed_handlers != 10) + { + usleep(100); + } + + registration::set_init_status(1); + registration::finalize(); +} \ No newline at end of file diff --git a/source/scripts/thread-sanitizer-suppr.txt b/source/scripts/thread-sanitizer-suppr.txt index 294fe51680..8c8f3f0d55 100644 --- a/source/scripts/thread-sanitizer-suppr.txt +++ b/source/scripts/thread-sanitizer-suppr.txt @@ -23,3 +23,6 @@ race:tzset_internal # bug in libtsan.so.0 which thinks there is a # double mutex lock (there isn't one) mutex:external/ptl/source/PTL/TaskGroup.hh + +# lock order inversion that cannot happen +mutex:source/lib/common/synchronized.hpp \ No newline at end of file