Skip to content

Remove callback interface from Queue #128

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
rocm-devops wants to merge 1 commit into
base: amd-staging
Choose a base branch
from
Open

Remove callback interface from Queue #128

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 1 addition & 5 deletions source/lib/rocprofiler-sdk/context/context.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,11 +76,7 @@ struct buffer_tracing_service

struct dispatch_counter_collection_service
{
// Contains a vector of counter collection instances associated with this context.
// Each instance is assocated with an agent and a counter collection profile.
// Contains callback information along with other data needed to collect/process
// counters.
std::vector<std::shared_ptr<counters::counter_callback_info>> callbacks{};
counters::counter_callback_info ctx_data{};
// A flag to state wether or not the counter set is currently enabled. This is primarily
// to protect against multithreaded calls to enable a context (and enabling already enabled
// counters).
Expand Down
19 changes: 9 additions & 10 deletions source/lib/rocprofiler-sdk/counters/controller.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
#include <rocprofiler-sdk/agent.h>
#include <rocprofiler-sdk/dispatch_counting_service.h>
#include <rocprofiler-sdk/fwd.h>
#include <optional>

namespace rocprofiler
{
Expand Down Expand Up @@ -149,20 +150,18 @@ CounterController::configure_dispatch(rocprofiler_context_id_t
std::make_unique<rocprofiler::context::dispatch_counter_collection_service>();
}

auto& cb =
*ctx.counter_collection->callbacks.emplace_back(std::make_shared<counter_callback_info>());
auto& ctx_data = ctx.counter_collection->ctx_data;

cb.user_cb = callback;
cb.callback_args = callback_args;
cb.context = context_id;
ctx_data.user_cb = callback;
ctx_data.callback_args = callback_args;
ctx_data.context = context_id;
if(buffer.handle != 0)
{
cb.buffer = buffer;
ctx_data.buffer = buffer;
}
cb.internal_context = ctx_p;
cb.record_callback = record_callback;
cb.record_callback_args = record_callback_args;

ctx_data.internal_context = ctx_p;
ctx_data.record_callback = record_callback;
ctx_data.record_callback_args = record_callback_args;
return ROCPROFILER_STATUS_SUCCESS;
}

Expand Down
34 changes: 0 additions & 34 deletions source/lib/rocprofiler-sdk/counters/core.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -160,40 +160,6 @@ start_context(const context::context* ctx)
if(!already_enabled)
{
callback_thread_start();

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,
const hsa::rocprofiler_packet& kern_pkt,
rocprofiler_kernel_id_t kernel_id,
rocprofiler_dispatch_id_t dispatch_id,
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(ctx,
cb,
q,
kern_pkt,
kernel_id,
dispatch_id,
user_data,
extern_corr_ids,
correlation_id);
},
// Completion CB
[=](const hsa::Queue& /* q */,
hsa::rocprofiler_packet /* kern_pkt */,
std::shared_ptr<hsa::Queue::queue_info_session_t>& session,
inst_pkt_t& aql,
kernel_dispatch::profiling_time dispatch_time) {
completed_cb(ctx, cb, session, aql, dispatch_time);
});
}
}
}

Expand Down
23 changes: 10 additions & 13 deletions source/lib/rocprofiler-sdk/counters/dispatch_handlers.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,7 @@ namespace counters
*/
hsa::Queue::pkt_and_serialize_t
queue_cb(const context::context* ctx,
const std::shared_ptr<counter_callback_info>& info,
counter_callback_info& info,
const hsa::Queue& queue,
const hsa::rocprofiler_packet& pkt,
rocprofiler_kernel_id_t kernel_id,
Expand All @@ -57,13 +57,11 @@ queue_cb(const context::context* ctx,
const hsa::Queue::queue_info_session_t::external_corr_id_map_t& extern_corr_ids,
const context::correlation_id* correlation_id)
{
CHECK(info && ctx);

// 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::EmptyAQLPacket>();
info->packet_return_map.wlock([&](auto& data) { data.emplace(ret_pkt.get(), nullptr); });
info.packet_return_map.wlock([&](auto& data) { data.emplace(ret_pkt.get(), 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.
Expand All @@ -77,15 +75,15 @@ queue_cb(const context::context* ctx,
ctx->counter_collection->enabled.rlock(
[&](const auto& collect_ctx) { is_enabled = collect_ctx; });

if(!is_enabled || !info->user_cb) return {no_instrumentation(), true};
if(!is_enabled || !info.user_cb) return {no_instrumentation(), true};

auto _corr_id_v =
rocprofiler_async_correlation_id_t{.internal = 0, .external = context::null_user_data};
if(const auto* _corr_id = correlation_id)
{
_corr_id_v.internal = _corr_id->internal;
if(const auto* external =
rocprofiler::common::get_val(extern_corr_ids, info->internal_context))
rocprofiler::common::get_val(extern_corr_ids, info.internal_context))
{
_corr_id_v.external = *external;
}
Expand Down Expand Up @@ -113,15 +111,15 @@ queue_cb(const context::context* ctx,
dispatch_data.dispatch_info = dispatch_info;
}

info->user_cb(dispatch_data, &req_profile, user_data, info->callback_args);
info.user_cb(dispatch_data, &req_profile, user_data, info.callback_args);

if(req_profile.handle == 0) return {no_instrumentation(), true};

auto prof_config = get_controller().get_profile_cfg(req_profile);
CHECK(prof_config);

std::unique_ptr<rocprofiler::hsa::AQLPacket> ret_pkt;
auto status = info->get_packet(ret_pkt, prof_config);
auto status = info.get_packet(ret_pkt, prof_config);
CHECK_EQ(status, ROCPROFILER_STATUS_SUCCESS) << rocprofiler_get_status_string(status);

if(!ret_pkt->empty)
Expand All @@ -140,17 +138,16 @@ queue_cb(const context::context* ctx,
*/
void
completed_cb(const context::context* ctx,
const std::shared_ptr<counter_callback_info>& info,
std::shared_ptr<hsa::Queue::queue_info_session_t>& ptr_session,
inst_pkt_t& pkts,
kernel_dispatch::profiling_time dispatch_time)
{
CHECK(info && ctx);

CHECK(ctx);
CHECK(ctx->counter_collection);
std::shared_ptr<counter_config> prof_config;
// Get the Profile Config
std::unique_ptr<rocprofiler::hsa::AQLPacket> pkt = nullptr;
info->packet_return_map.wlock([&](auto& data) {
ctx->counter_collection->ctx_data.packet_return_map.wlock([&](auto& data) {
for(auto& [aql_pkt, _] : pkts)
{
const auto* profile = rocprofiler::common::get_val(data, aql_pkt.get());
Expand All @@ -167,7 +164,7 @@ completed_cb(const context::context* ctx,
// We have no profile config, nothing to output.
if(!pkt || !prof_config) return;

completed_cb_params_t params{info, ptr_session, dispatch_time, prof_config, std::move(pkt)};
completed_cb_params_t params{ctx, ptr_session, dispatch_time, prof_config, std::move(pkt)};
process_callback_data(std::move(params));
}

Expand Down
3 changes: 1 addition & 2 deletions source/lib/rocprofiler-sdk/counters/dispatch_handlers.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@ using inst_pkt_t = common::container::

hsa::Queue::pkt_and_serialize_t
queue_cb(const context::context* ctx,
const std::shared_ptr<counter_callback_info>& info,
counter_callback_info& info,
const hsa::Queue& queue,
const hsa::rocprofiler_packet& pkt,
rocprofiler_kernel_id_t kernel_id,
Expand All @@ -47,7 +47,6 @@ queue_cb(const context::context* ctx,

void
completed_cb(const context::context* ctx,
const std::shared_ptr<counter_callback_info>& info,
std::shared_ptr<hsa::Queue::queue_info_session_t>& session,
inst_pkt_t& pkts,
kernel_dispatch::profiling_time dispatch_time);
Expand Down
22 changes: 11 additions & 11 deletions source/lib/rocprofiler-sdk/counters/sample_processing.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,8 @@ get_buffer_mut()
void
proccess_completed_cb(completed_cb_params_t&& params)
{
auto& info = params.info;
CHECK(params.ctx && params.ctx->counter_collection);
auto& info = params.ctx->counter_collection->ctx_data;
auto& session = *params.session;
auto& dispatch_time = params.dispatch_time;
auto& prof_config = params.prof_config;
Expand All @@ -70,9 +71,9 @@ proccess_completed_cb(completed_cb_params_t&& params)
common::container::small_vector<rocprofiler_counter_record_t, 128> out;
rocprofiler::buffer::instance* buf = nullptr;

if(info->buffer)
if(info.buffer)
{
buf = CHECK_NOTNULL(buffer::get_buffer(info->buffer->handle));
buf = CHECK_NOTNULL(buffer::get_buffer(info.buffer->handle));
}

auto _corr_id_v =
Expand All @@ -82,7 +83,7 @@ proccess_completed_cb(completed_cb_params_t&& params)
{
_corr_id_v.internal = _corr_id->internal;
if(const auto* external = rocprofiler::common::get_val(
session.tracing_data.external_correlation_ids, info->internal_context))
session.tracing_data.external_correlation_ids, info.internal_context))
{
_corr_id_v.external = *external;
}
Expand Down Expand Up @@ -132,8 +133,7 @@ proccess_completed_cb(completed_cb_params_t&& params)
}
else
{
CHECK(info->record_callback);

CHECK(info.record_callback);
auto dispatch_data =
common::init_public_api_struct(rocprofiler_dispatch_counting_service_data_t{});

Expand All @@ -145,11 +145,11 @@ proccess_completed_cb(completed_cb_params_t&& params)
dispatch_data.end_timestamp = dispatch_time.end;
}

info->record_callback(dispatch_data,
out.data(),
out.size(),
session.user_data,
info->record_callback_args);
info.record_callback(dispatch_data,
out.data(),
out.size(),
session.user_data,
info.record_callback_args);
}
}
}
Expand Down
2 changes: 1 addition & 1 deletion source/lib/rocprofiler-sdk/counters/sample_processing.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ namespace counters
{
struct completed_cb_params_t
{
std::shared_ptr<counter_callback_info> info;
const context::context* ctx;
std::shared_ptr<hsa::Queue::queue_info_session_t> session;
kernel_dispatch::profiling_time dispatch_time;
std::shared_ptr<counter_config> prof_config;
Expand Down
52 changes: 16 additions & 36 deletions source/lib/rocprofiler-sdk/counters/tests/core.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -407,10 +407,10 @@ TEST(core, check_callbacks)
auto profile = counters::get_counter_config(expected.id);
ASSERT_TRUE(profile);

std::shared_ptr<counters::counter_callback_info> cb_info =
std::make_shared<counters::counter_callback_info>();
cb_info->user_cb = user_dispatch_cb;
cb_info->callback_args = static_cast<void*>(&expected);
auto& cb_info = ctx.counter_collection->ctx_data;

cb_info.user_cb = user_dispatch_cb;
cb_info.callback_args = static_cast<void*>(&expected);

context::correlation_id corr_id;
corr_id.internal = count++;
Expand Down Expand Up @@ -460,7 +460,7 @@ TEST(core, check_callbacks)
nullptr,
&opt_buff_id),
"Could not create buffer");
cb_info->buffer = opt_buff_id;
cb_info.buffer = opt_buff_id;

auto _sess = hsa::Queue::queue_info_session_t{.queue = fq};
_sess.correlation_id = &corr_id;
Expand All @@ -470,7 +470,7 @@ TEST(core, check_callbacks)
counters::inst_pkt_t pkts;
pkts.emplace_back(
std::make_pair(std::move(ret_pkt.pkt), static_cast<counters::ClientID>(0)));
completed_cb(&ctx, cb_info, sess, pkts, kernel_dispatch::profiling_time{});
completed_cb(&ctx, sess, pkts, kernel_dispatch::profiling_time{});
rocprofiler_flush_buffer(opt_buff_id);
rocprofiler_destroy_buffer(opt_buff_id);
}
Expand Down Expand Up @@ -551,26 +551,16 @@ TEST(core, start_stop_buffered_ctx)
auto& ctx = *ctx_p;

ASSERT_TRUE(ctx.counter_collection);
ASSERT_EQ(ctx.counter_collection->callbacks.size(), 1);
EXPECT_EQ(ctx.counter_collection->callbacks.at(0)->user_cb, null_dispatch_callback);
EXPECT_EQ(ctx.counter_collection->callbacks.at(0)->callback_args, (void*) 0x12345);
EXPECT_EQ(ctx.counter_collection->callbacks.at(0)->context.handle, get_client_ctx().handle);
ASSERT_TRUE(ctx.counter_collection->callbacks.at(0)->buffer);
EXPECT_EQ(ctx.counter_collection->callbacks.at(0)->buffer->handle, opt_buff_id.handle);
EXPECT_EQ(ctx.counter_collection->ctx_data.user_cb, null_dispatch_callback);
EXPECT_EQ(ctx.counter_collection->ctx_data.callback_args, (void*) 0x12345);
EXPECT_EQ(ctx.counter_collection->ctx_data.context.handle, get_client_ctx().handle);
ASSERT_TRUE(ctx.counter_collection->ctx_data.buffer);
EXPECT_EQ(ctx.counter_collection->ctx_data.buffer->handle, opt_buff_id.handle);

bool found = false;
ctx.counter_collection->enabled.rlock([&](const auto& data) { found = data; });
EXPECT_TRUE(found);

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_TRUE(found);

/**
* Check if context can be disabled correctly
*/
Expand Down Expand Up @@ -616,26 +606,16 @@ TEST(core, start_stop_callback_ctx)
auto& ctx = *ctx_p;

ASSERT_TRUE(ctx.counter_collection);
ASSERT_EQ(ctx.counter_collection->callbacks.size(), 1);
EXPECT_EQ(ctx.counter_collection->callbacks.at(0)->user_cb, null_dispatch_callback);
EXPECT_EQ(ctx.counter_collection->callbacks.at(0)->callback_args, (void*) 0x12345);
EXPECT_EQ(ctx.counter_collection->callbacks.at(0)->record_callback, null_record_callback);
EXPECT_EQ(ctx.counter_collection->callbacks.at(0)->record_callback_args, (void*) 0x54321);
EXPECT_EQ(ctx.counter_collection->callbacks.at(0)->context.handle, get_client_ctx().handle);
EXPECT_EQ(ctx.counter_collection->ctx_data.user_cb, null_dispatch_callback);
EXPECT_EQ(ctx.counter_collection->ctx_data.callback_args, (void*) 0x12345);
EXPECT_EQ(ctx.counter_collection->ctx_data.record_callback, null_record_callback);
EXPECT_EQ(ctx.counter_collection->ctx_data.record_callback_args, (void*) 0x54321);
EXPECT_EQ(ctx.counter_collection->ctx_data.context.handle, get_client_ctx().handle);

bool found = false;
ctx.counter_collection->enabled.rlock([&](const auto& data) { found = data; });
EXPECT_TRUE(found);

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_TRUE(found);

/**
* Check if context can be disabled correctly
*/
Expand Down
Loading