AI/rocm-systems
2
0
Fork 0
程式碼 問題 合併請求 Actions 套件 專案 版本發布 Wiki 動態

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>
此提交包含在：
Benjamin Welton
2024-03-08 07:02:43 -08:00
提交者 GitHub
父節點 7b6d3c70bd
當前提交 1de44447f4
共有 16 個檔案被更改，包括 1099 行新增 和 234 行删除
下載 Patch 檔 下載差異檔 展開所有檔案 折疊所有檔案
source/lib/common/synchronized.hpp
-1
查看文件
@@ -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 <typename FuncT, typename... Args>
decltype(auto) rlock(FuncT&& lambda, Args&&... args) const;
source/lib/rocprofiler-sdk/counters/core.cpp
+76 -40
查看文件
@@ -253,7 +253,8 @@ counter_callback_info::get_packet(std::unique_ptr<rocprofiler::hsa::AQLPacket>&
* We return an AQLPacket containing the start/stop/read packets for injection.
*/
std::unique_ptr<rocprofiler::hsa::AQLPacket>
queue_cb(const std::shared_ptr<counter_callback_info>& info,
queue_cb(const context::context* ctx,
const std::shared_ptr<counter_callback_info>& info,
const hsa::Queue& queue,
const hsa::rocprofiler_packet& pkt,
uint64_t kernel_id,
@@ -261,7 +262,43 @@ queue_cb(const std::shared_ptr<counter_callback_info>& 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<rocprofiler::hsa::AQLPacket>(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<counter_callback_info>& 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<counter_callback_info>& 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<hsa_ext_amd_aql_pm4_packet_t, 3>& _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<counter_callback_info>& info,
* Callback called by HSA interceptor when the kernel has completed processing.
*/
void
completed_cb(const std::shared_ptr<counter_callback_info>& info,
completed_cb(const context::context* ctx,
const std::shared_ptr<counter_callback_info>& 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<profile_config> prof_config;
// Get the Profile Config
@@ -365,10 +392,11 @@ completed_cb(const std::shared_ptr<counter_callback_info>& 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
source/lib/rocprofiler-sdk/counters/core.hpp
+7 -5
查看文件
@@ -126,7 +126,8 @@ void
stop_context(const context::context*);
std::unique_ptr<rocprofiler::hsa::AQLPacket>
queue_cb(const std::shared_ptr<counter_callback_info>& info,
queue_cb(const context::context* ctx,
const std::shared_ptr<counter_callback_info>& 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<std::pair<std::unique_ptr<rocprofiler::hsa::AQLPacket>, ClientID>, 4>;
void
completed_cb(const std::shared_ptr<counter_callback_info>&,
const hsa::Queue&,
completed_cb(const context::context* ctx,
const std::shared_ptr<counter_callback_info>& 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<profile_config> get_profile_config(rocprofiler_profile_config_id_t);
} // namespace counters
source/lib/rocprofiler-sdk/counters/tests/core.cpp
+7 -18
查看文件
@@ -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<rocprofiler::context::counter_collection_service>();
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<counters::ClientID>(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);
source/lib/rocprofiler-sdk/hsa/CMakeLists.txt
+12 -2
查看文件
@@ -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
source/lib/rocprofiler-sdk/hsa/hsa_barrier.cpp
+111
查看文件
@@ -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<rocprofiler_packet>
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
source/lib/rocprofiler-sdk/hsa/hsa_barrier.hpp
+84
查看文件
@@ -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 <functional>
#include <optional>
#include <unordered_map>
#include <unordered_set>
#include <hsa/hsa.h>
#include <hsa/hsa_api_trace.h>
#include <hsa/hsa_ext_amd.h>
#include <rocprofiler-sdk/rocprofiler.h>
#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_queue_t*, std::unique_ptr<Queue>>;
hsa_barrier(std::function<void()>&& 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<rocprofiler_packet> 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<void()> _barried_finished;
CoreApiTable _core_api;
common::Synchronized<std::unordered_map<int64_t, int64_t>> _queue_waiting;
common::Synchronized<std::unordered_set<int64_t>> _barrier_enqueued;
std::atomic<bool> _complete{false};
// Blocks all queues from executing until the barrier is lifted
hsa_signal_t _barrier_signal{};
};
} // namespace hsa
} // namespace rocprofiler
source/lib/rocprofiler-sdk/hsa/profile_serializer.cpp
+247
查看文件
@@ -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<hsa_queue_t*>(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<profiler_serializer::barrier_with_state>& 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<std::mutex> 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<hsa::rocprofiler_packet, 3>
profiler_serializer::kernel_dispatch(const Queue& queue) const
{
common::container::small_vector<hsa::rocprofiler_packet, 3> 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<hsa_barrier>([] {}, 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<hsa_barrier>([] {}, get_queue_controller().get_core_table()));
_serializer_status = Status::DISABLED;
_barrier.back().barrier->set_barrier(queues);
}
} // namespace hsa
} // namespace rocprofiler
source/lib/rocprofiler-sdk/hsa/profile_serializer.hpp
+98
查看文件
@@ -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 <rocprofiler-sdk/rocprofiler.h>
#include "lib/common/container/small_vector.hpp"
#include "lib/rocprofiler-sdk/hsa/hsa_barrier.hpp"
#include "lib/rocprofiler-sdk/hsa/queue.hpp"
#include <cstdint>
#include <functional>
#include <optional>
#include <unordered_map>
#include <vector>
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<hsa_barrier> _barrier)
: state(_state)
, barrier(std::move(_barrier))
{}
Status state;
std::unique_ptr<hsa_barrier> barrier;
};
using queue_map_t = std::unordered_map<hsa_queue_t*, std::unique_ptr<Queue>>;
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<hsa::rocprofiler_packet, 3> 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<const Queue*> _dispatch_ready;
std::atomic<Status> _serializer_status{Status::DISABLED};
std::deque<barrier_with_state> _barrier;
};
} // namespace hsa
} // namespace rocprofiler
source/lib/rocprofiler-sdk/hsa/queue.cpp
+22 -7
查看文件
@@ -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 <glog/logging.h>
#include <hsa/hsa.h>
@@ -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<rocprofiler_packet>& _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<void*>(data));
signal, HSA_SIGNAL_CONDITION_EQ, -1, AsyncSignalHandler, static_cast<void*>(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;
}
source/lib/rocprofiler-sdk/hsa/queue.hpp
+16 -3
查看文件
@@ -173,6 +173,9 @@ public:
template <typename FuncT>
void signal_callback(FuncT&& func) const;
template <typename FuncT>
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<callback_map_t> _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 <typename FuncT>
void
Queue::lock_queue(FuncT&& func)
{
std::unique_lock<std::mutex> lock(_lock_queue);
func();
}
} // namespace hsa
} // namespace rocprofiler
source/lib/rocprofiler-sdk/hsa/queue_controller.cpp
+34 -124
查看文件
@@ -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<std::mutex> 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 <typename FuncT>
void
QueueController::profiler_serializer(FuncT&& lambda)
{
_profiler_serializer.wlock(std::forward<FuncT>(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<hsa_queue_t*>(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<std::mutex> 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
{
source/lib/rocprofiler-sdk/hsa/queue_controller.hpp
+23 -33
查看文件
@@ -24,6 +24,7 @@
#include <rocprofiler-sdk/rocprofiler.h>
#include "lib/rocprofiler-sdk/hsa/profile_serializer.hpp"
#include "lib/rocprofiler-sdk/hsa/queue.hpp"
#include <cstdint>
@@ -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<const Queue*> dispatch_ready;
bool enabled{true};
};
// Tracks and manages HSA queues
class QueueController
{
@@ -65,6 +45,7 @@ public:
std::tuple<rocprofiler_agent_t, Queue::queue_cb_t, Queue::completed_cb_t>;
using queue_iterator_cb_t = std::function<void(const Queue*)>;
using callback_iterator_cb_t = std::function<void(ClientID, const agent_callback_tuple_t&)>;
using queue_map_t = std::unordered_map<hsa_queue_t*, std::unique_ptr<Queue>>;
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 <typename FuncT>
void profiler_serializer(FuncT&& lambda);
void iterate_callbacks(const callback_iterator_cb_t&) const;
common::Synchronized<hsa::profiler_serializer>& 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<std::unordered_map<uint64_t, hsa_signal_t>> _debug_signals;
#endif
private:
using queue_map_t = std::unordered_map<hsa_queue_t*, std::unique_ptr<Queue>>;
using client_id_map_t = std::unordered_map<ClientID, agent_callback_tuple_t>;
using agent_cache_map_t = std::unordered_map<uint32_t, AgentCache>;
CoreApiTable _core_table = {};
AmdExtTable _ext_table = {};
common::Synchronized<queue_map_t> _queues = {};
common::Synchronized<client_id_map_t> _callback_cache = {};
agent_cache_map_t _supported_agents = {};
common::Synchronized<profiler_serializer_t> _profiler_serializer;
CoreApiTable _core_table = {};
AmdExtTable _ext_table = {};
common::Synchronized<queue_map_t> _queues = {};
common::Synchronized<client_id_map_t> _callback_cache = {};
agent_cache_map_t _supported_agents = {};
common::Synchronized<hsa::profiler_serializer> _profiler_serializer;
};
QueueController&
source/lib/rocprofiler-sdk/tests/CMakeLists.txt
+1 -1
查看文件
@@ -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)
source/lib/rocprofiler-sdk/tests/hsa_barrier.cpp
+358
查看文件
@@ -0,0 +1,358 @@
#include <algorithm>
#include <random>
#include <rocprofiler-sdk/agent.h>
#include <rocprofiler-sdk/buffer.h>
#include <rocprofiler-sdk/fwd.h>
#include <rocprofiler-sdk/registration.h>
#include <gtest/gtest.h>
#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<FakeQueue>(agent_cache, rocprofiler_queue_id_t{.handle = i});
}
return ret;
}
std::atomic<bool> should_execute_handler{false};
std::atomic<int> 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<hsa_signal_t*>(data));
delete static_cast<hsa_signal_t*>(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<void*>(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<Queue*> 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<Queue*> 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();
}
source/scripts/thread-sanitizer-suppr.txt
+3
查看文件
@@ -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