diff --git a/projects/rocprofiler-sdk/source/docs/conceptual/queue_hooks/queue_callback_removal_architecture.md b/projects/rocprofiler-sdk/source/docs/conceptual/queue_hooks/queue_callback_removal_architecture.md new file mode 100644 index 000000000000..7edc81710e04 --- /dev/null +++ b/projects/rocprofiler-sdk/source/docs/conceptual/queue_hooks/queue_callback_removal_architecture.md @@ -0,0 +1,208 @@ +# Queue Callback Registry Removal — Software Architecture + +How queue-interposed services are wired into dispatch interception after the per-queue callback +registry is removed, why completion routing and enqueue routing deliberately use different context +sets, and why `stop_context` drains the GPU. + +Paths are relative to `projects/rocprofiler-sdk/source/`. Symbols are named rather than cited by +line number, since line numbers rot faster than the code they point at. + +The migration is one PR per service. Counter collection (#8891) introduces the shared +`hsa/queue_hooks/` infrastructure that the rest build on; SPM (#8887), thread trace (#8790) and PC +sampling (#8895) follow the same shape. This document describes the mechanism as a whole, and flags +which service each part currently lives on. + +## 1. Diagram + +```mermaid +flowchart TB + + subgraph BEFORE["BEFORE — per-queue callback registry"] + direction TB + B1["service start_context"] + B2["QueueController::add_callback
auto-incrementing ClientID"] + B3["Queue::_callbacks map
ClientID -> queue_callbacks_t
_notifiers counter"] + B4["WriteInterceptor
queue.signal_callback(...)
iterate map, call write_interceptor"] + B5["AsyncSignalHandler
queue.signal_callback(...)
iterate map, call completed_cb"] + B6["captured context + callback refs
routing by provenance:
a registered callback fires even
after its context is stopped"] + B1 --> B2 --> B3 + B3 --> B4 + B3 --> B5 + B3 --- B6 + end + + subgraph AFTER["AFTER — explicit hooks"] + direction TB + C1["service start_context
no registration"] + C2["WriteInterceptor
counters::kernel_dispatch_phase_enter_hook
called inline, unconditionally"] + C3["counters::is_any_active
folded into no_real_consumers and
forces should_batch_packets = false"] + C4["AsyncSignalHandler
counters::kernel_dispatch_phase_exit_hook
called inline"] + C5["hsa/queue_hooks/client_ids.hpp
stable producer tags
COUNTERS / SPM / THREAD_TRACE"] + C1 --> C2 + C2 --> C3 + C1 --> C4 + C2 --- C5 + C4 --- C5 + end + + subgraph ROUTE["ROUTING — the two context sets are not interchangeable"] + direction TB + R1["enter hook: get_active_contexts
new instrumentation must stop
as soon as the context stops"] + R2["exit hook: get_registered_contexts
work already on the GPU must
still be able to complete"] + R3["using active contexts in the exit hook
drops the completion of any dispatch
in flight at stop_context:
record never delivered,
packet_return_map entry leaked"] + R2 -.->|"if this were
get_active_contexts"| R3 + end + + subgraph STOP["STOP PATH — ordering is load-bearing"] + direction TB + S1["context::stop_context stops
queue-interposed services before
clearing the active slot"] + S2["counters::stop_context
1. clear the enabled flag
2. queue_controller_sync (drain)
3. disable_serialization
4. callback_thread_stop"] + S3["the service stays visible for the
whole drain, so queue_cb's disabled
path keeps returning serialize=true
and no separate draining flag is needed"] + S1 --> S2 --> S3 + end + + BEFORE ==> AFTER + AFTER ==> ROUTE + ROUTE ==> STOP + + classDef old fill:#eceff1,stroke:#546e7a,stroke-width:2px,color:#000 + classDef new fill:#e8f5e9,stroke:#2e7d32,stroke-width:2px,color:#000 + classDef bad fill:#ffebee,stroke:#c62828,stroke-width:2px,color:#000 + classDef fix fill:#e3f2fd,stroke:#1565c0,stroke-width:2px,color:#000 + class B1,B2,B3,B4,B5,B6 old + class C1,C2,C3,C4,C5 new + class R3 bad + class R1,R2,S1,S2,S3 fix +``` + +## 2. What the migration changes + +Before, a service registered a `queue_callbacks_t` pair with the queue controller when its context +started. The write interceptor and the async signal handler both walked `Queue::_callbacks` and +invoked whatever was registered, and `Queue::get_notifiers()` counted the registrations so the +interceptor could early-out when nothing was subscribed. + +After, the service registers nothing. The write interceptor and the async signal handler call the +service's hooks directly, and each hook decides for itself whether it has work to do by querying the +context list. + +For counter collection: + +| Element | Location | +|---|---| +| `counters::kernel_dispatch_phase_enter_hook` | `counters/queue_hooks.hpp`, `counters/queue_hooks.cpp` | +| `counters::kernel_dispatch_phase_exit_hook` | `counters/queue_hooks.hpp`, `counters/queue_hooks.cpp` | +| `counters::is_any_active` | `counters/queue_hooks.hpp`, `counters/queue_hooks.cpp` | +| Context filter shared by all three | `counters/queue_hooks.cpp`, `counter_contexts_filter()` | +| Stable producer tags | `hsa/queue_hooks/client_ids.hpp` | +| Exit hook call site | `hsa/queue.cpp`, in the async signal handler | +| `no_real_consumers` gains `!counters::is_any_active()` | `hsa/queue.cpp` | +| Enter hook call site | `hsa/queue.cpp`, in the write interceptor | +| Batching disabled while counters are active | `hsa/queue.cpp`, `should_batch_packets` | +| Service stop path | `counters/core.cpp`, `stop_context()` | + +The hook names describe the dispatch phase they run in: the enter hook runs when a dispatch is being +submitted, the exit hook when its completion signal fires. `is_any_active` is neither phase, so it +keeps a plain name. Each remaining service mirrors this trio in its own `queue_hooks.{hpp,cpp}`. + +`client_ids.hpp` replaces the registry's auto-incrementing `ClientID` with fixed producer tags, so +the id attached to an instrumentation packet no longer depends on the order in which services +happened to register. The tags numerically overlap the registry's `ClientID`, which also starts at 1 +and remains in use by services that have not migrated yet. That is inert only because no consumer +dispatches on the tag — each service identifies its own packets by pointer lookup or `dynamic_cast` +— and `client_ids.hpp` records what has to change before anything routes on these values. + +## 3. Enqueue and completion use different context sets + +This is the central design point, and getting it wrong is a silent data-loss bug rather than a crash. + +The registry routed completions by **provenance**: the callback pair captured at enqueue time was +invoked when the packet completed, regardless of whether the owning context was still active. The +hooks have to reproduce that property without the registry, and they do it by choosing different +context sets for the two phases: + +- The **enter hook** iterates `get_active_contexts`. New instrumentation must stop being added as + soon as the context stops. +- The **exit hook** iterates `get_registered_contexts`. A dispatch already executing on the GPU must + still be able to deliver its record, even though its context is no longer active. + +Routing the exit hook over active contexts instead loses any dispatch that is in flight when its +context is stopped: `completed_cb` never runs, so the record is never delivered to the tool, and the +`packet_return_map` entry is never erased, leaking the AQL packet and its profile allocation. Because +pause/resume is implemented as `stop_context`/`start_context`, this shows up as missing counter +records around every pause. + +Iterating registered contexts is safe because `completed_cb` self-filters: it resolves ownership by +looking the packet up in its own context's `packet_return_map` and returns early for packets it does +not own. Each context therefore still processes only its own packets, and the guarantee is restored +without reintroducing a registry. + +## 4. The stop path + +Two orderings matter, both stated at their call sites in the code. + +**`context::stop_context` stops queue-interposed services before clearing the active slot.** Counter +collection, SPM, device thread trace and dispatch thread trace are all stopped ahead of the +`compare_exchange_strong` that nulls the context's active slot. Clearing the slot first opens a +window in which the enter hook sees no active context, so a dispatch is submitted without serializer +packets while the serializer is still enabled. + +**`counters::stop_context` drains the GPU before disabling serialization.** The order is: clear the +service's `enabled` flag, `hsa::queue_controller_sync()`, `disable_serialization()`, then +`callback_thread_stop()`. + +### 4.1 Why the drain is kept + +Provenance routing and the drain answer different questions, so one does not replace the other. +Provenance routing guarantees that a completion which *arrives* is delivered. The drain bounds +*when completions arrive at all*, which is what the teardown downstream of it depends on: it is what +makes "the callback thread and the `counter_callback_info` objects outlive every in-flight dispatch" +true literally, rather than true by an argument about what happens if they do not. + +The supporting facts are worth recording, because they are the reason a missing drain is hard to +notice rather than a reason to omit it: + +| Property | Why it holds | +|---|---| +| `counter_callback_info` lifetime | Structural. The context holds them as `std::vector>` on the service, and neither `counters::stop_context` nor `context::stop_context` clears that vector. Since the exit hook iterates *registered* contexts, the context and its callbacks are still reachable. | +| Callback thread does not strand queued work | `consumer_thread_t::exit()` clears `valid` then waits on `exited`, and `consumer_loop` only sets `exited` once `read_ptr == write_ptr`, so the queue drains before the join. Afterwards `consumer_thread_t::add()` takes the self-consume path and runs inline on the caller. Covered by `counters/tests/consumer_test.cpp`, `restart`. | +| Serializer transition with in-flight serialized dispatches | GPU-ordered independently. `profiler_serializer::disable()` records the previous state and pushes an `hsa_barrier` across the queues, and `kernel_completion_signal` reconciles in-flight dispatches against it. | +| No separate "draining" flag is needed | `context::stop_context` calls the service stop path while the context is still in the active list, so throughout the drain the enter hook still reaches `queue_cb`, whose disabled path returns `serialize=true` and keeps the serialized-to-unserialized transition coordinated. This only works because the drain happens before the slot is cleared. | + +The drain is a bound, not a hard barrier: `Queue::sync` uses a five-second hint and only warns on +timeout, and `_active_kernels` counts intercepted dispatches only. It is the ordering guarantee for +teardown, and provenance routing in the exit hook is what makes individual completions correct. + +One hazard is not covered by either mechanism: the drain incidentally protects the sequence "stop the +context, then destroy the counter config". If a tool does that, the guard belongs at the destroy +path as a drain or a refcount, since that is where the lifetime actually ends. + +## 5. Relationship to kernel replay (#8622) + +Callback removal is a soft prerequisite for kernel replay, not a blocker, and the two can land in +either order. + +It helps because with the registry there is no explicit "if this context is active" block in the +write interceptor — the decision is hidden inside the registered callbacks, so disabling a service +for a single replay pass means mutating the callback structure. Once the hooks are inline, the +interceptor has an explicit conditional on context activity, which is a much easier place to express +that a context is active globally but inactive for this pass. + +The routing rule in section 3 matters more under replay than without it. Replay runs N passes per +dispatch and reuses `process_packet_batch` for each pass, so every pass creates its own +instrumentation packet and its own `packet_return_map` entry. A dropped completion is one lost record +without replay; under replay it is N leaked packets per dispatch plus silently missing counter +groups. The replay loop also waits on its `pass_done` barrier with no timeout, so a stall on the +completion path becomes an application hang rather than data loss. + +## 6. Known gaps + +1. Thread trace, PC sampling and device counter collection still use the registry, so + `Queue::_callbacks`, `Queue::get_notifiers()` and `QueueController::add_callback` cannot be + deleted yet. `client_ids.hpp` already reserves tags for thread trace and PC sampling. +2. Several in-flight PRs edit the same `no_real_consumers` expression in `hsa/queue.cpp`. + Consolidating the predicate into one `needs_interception(queue)` helper would remove the + recurring conflict. +3. `counters::is_any_active()` forces `should_batch_packets = false` for every dispatch while any + counter context is active. The performance effect of that, combined with kernel replay's own + gate, has not been measured. diff --git a/projects/rocprofiler-sdk/source/include/rocprofiler-sdk/dispatch_counting_service.h b/projects/rocprofiler-sdk/source/include/rocprofiler-sdk/dispatch_counting_service.h index 2b1e524e8c5a..87bdafef8df5 100644 --- a/projects/rocprofiler-sdk/source/include/rocprofiler-sdk/dispatch_counting_service.h +++ b/projects/rocprofiler-sdk/source/include/rocprofiler-sdk/dispatch_counting_service.h @@ -110,20 +110,6 @@ typedef void (*rocprofiler_dispatch_counting_record_cb_t)( * one dispatch (denoted by correlation id). Will trigger the * callback based on the parameters setup in buffer_id_t. * - * NOTE: Interface is up for comment as to whether restrictions - * on agent should be made here (limiting the CB based on agent) - * or if the restriction should be performed by the tool in - * ::rocprofiler_dispatch_counting_service_cb_t (i.e. - * tool code checking the agent param to see if they want to profile - * it). - * - * Interface is up for comment as to whether restrictions - * on agent should be made here (limiting the CB based on agent) - * or if the restriction should be performed by the tool in - * ::rocprofiler_dispatch_counting_service_cb_t (i.e. - * tool code checking the agent param to see if they want to profile - * it). - * * @param [in] context_id context id * @param [in] buffer_id id of the buffer to use for the counting service * @param [in] callback callback to perform when dispatch is enqueued @@ -159,6 +145,45 @@ rocprofiler_configure_callback_dispatch_counting_service( rocprofiler_dispatch_counting_record_cb_t record_callback, void* record_callback_args) ROCPROFILER_API; +/** + * @brief (experimental) Restrict a dispatch counting service to a set of GPU agents. + * + * By default a dispatch counting service applies to every GPU agent: the dispatch + * callback is invoked for kernels on all agents, and kernel dispatches are serialized + * on all agents for as long as the context is active. A tool can already decline to + * profile a dispatch by leaving the ::rocprofiler_counter_config_id_t output of + * ::rocprofiler_dispatch_counting_service_cb_t set to zero, but that only suppresses + * counter collection -- the dispatch is still serialized. This function narrows the + * service so that agents outside @p agents are neither instrumented nor serialized. + * + * Restricting the agent set also relaxes the counter-collection context conflict: + * two contexts may each configure a dispatch counting service and be active at the + * same time provided their agent sets are disjoint. Contexts that leave the agent set + * unrestricted claim every agent and therefore still conflict with each other. + * + * Must be called after a dispatch counting service has been configured on + * @p context_id and before the context is started. Calling it more than once replaces + * the previous set. Passing @p num_agents of zero restores the default of all agents. + * + * @param [in] context_id context id with a configured dispatch counting service + * @param [in] agents array of GPU agent ids to restrict collection to + * @param [in] num_agents number of entries in @p agents + * @retval ::ROCPROFILER_STATUS_SUCCESS agent set was applied + * @retval ::ROCPROFILER_STATUS_ERROR_CONTEXT_INVALID @p context_id is not a valid context + * @retval ::ROCPROFILER_STATUS_ERROR_CONTEXT_NOT_FOUND no dispatch counting service has been + * configured on @p context_id + * @retval ::ROCPROFILER_STATUS_ERROR_INVALID_ARGUMENT @p agents is null with a non-zero + * @p num_agents + * @retval ::ROCPROFILER_STATUS_ERROR_AGENT_NOT_FOUND an entry in @p agents is not a GPU agent + * @retval ::ROCPROFILER_STATUS_ERROR_CONFIGURATION_LOCKED the context is currently active + * @return ::rocprofiler_status_t + */ +ROCPROFILER_SDK_EXPERIMENTAL +rocprofiler_status_t +rocprofiler_dispatch_counting_service_set_agents(rocprofiler_context_id_t context_id, + const rocprofiler_agent_id_t* agents, + size_t num_agents) ROCPROFILER_API; + /** @} */ ROCPROFILER_EXTERN_C_FINI diff --git a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/context/context.cpp b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/context/context.cpp index bf8d0c92299f..b8ae91ec2833 100644 --- a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/context/context.cpp +++ b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/context/context.cpp @@ -23,6 +23,7 @@ #include "lib/rocprofiler-sdk/context/context.hpp" #include "lib/common/container/small_vector.hpp" #include "lib/common/container/stable_vector.hpp" +#include "lib/common/scope_destructor.hpp" #include "lib/common/static_object.hpp" #include "lib/common/synchronized.hpp" #include "lib/common/utility.hpp" @@ -38,33 +39,41 @@ #include #include +#include #include #include #include #include #include -#include #include +#include +#include +#include namespace rocprofiler { namespace context { +// Owns the contexts it names. A reader holding a snapshot keeps them alive for the duration of +// the read, so deregister_client_contexts() dropping the registry's reference cannot destroy a +// context out from under an in-flight completion. The parallel raw-pointer array is what the +// snapshot iterates, so walking one costs no reference counting. +struct registered_context_storage +{ + std::vector> owned = {}; + std::vector view = {}; +}; + namespace { -using reserve_size_t = common::container::reserve_size; -using stable_context_vec_t = common::container::stable_vector, 8>; -using active_context_vec_t = common::container::stable_vector, 8>; +using reserve_size_t = common::container::reserve_size; +using context_ptr_t = std::shared_ptr; +using stable_context_vec_t = common::container::stable_vector; +using active_context_vec_t = common::container::stable_vector, 8>; +using context_snapshot_ptr_t = std::shared_ptr; constexpr auto invalid_client_idx = std::numeric_limits::max(); -auto& -get_contexts_mutex() -{ - static auto _v = std::mutex{}; - return _v; -} - uint64_t get_contexts_offset() { @@ -84,14 +93,6 @@ get_client_index() return _v; } -stable_context_vec_t*& -get_registered_contexts_impl() -{ - static auto*& _v = common::static_object::construct( - reserve_size_t{stable_context_vec_t::chunk_size}); - return _v; -} - auto& get_num_active_contexts() { @@ -105,26 +106,199 @@ get_active_contexts_impl() static auto* _v = new active_context_vec_t{reserve_size_t{active_context_vec_t::chunk_size}}; return *_v; } + +// C++20 replaces these with std::atomic>; the free functions are the C++17 +// spelling and are deprecated there, so select on the feature macro rather than leaving a +// deprecation warning to trip -Werror on a future standard bump. +#if defined(__cpp_lib_atomic_shared_ptr) && __cpp_lib_atomic_shared_ptr >= 201711L +using published_snapshot_t = std::atomic; + +context_snapshot_ptr_t +load_snapshot(published_snapshot_t& _v) +{ + return _v.load(std::memory_order_acquire); +} + +void +store_snapshot(published_snapshot_t& _v, context_snapshot_ptr_t&& _new) +{ + _v.store(std::move(_new), std::memory_order_release); +} +#else +using published_snapshot_t = context_snapshot_ptr_t; + +context_snapshot_ptr_t +load_snapshot(published_snapshot_t& _v) +{ + return std::atomic_load_explicit(&_v, std::memory_order_acquire); +} + +void +store_snapshot(published_snapshot_t& _v, context_snapshot_ptr_t&& _new) +{ + std::atomic_store_explicit(&_v, std::move(_new), std::memory_order_release); +} +#endif + +// The registry and the snapshot published from it live in one static object rather than two. +// common::destroy_static_objects() tears static objects down in reverse construction order, so two +// objects would be ordered by whichever happened to be touched first -- and if the snapshot +// outlived the registry it would be holding the last reference to every context, destroying them +// after the HSA tables that a context destructor calls into are already gone. As members, the +// snapshot is destroyed before the registry by the ordinary reverse-member rule. +struct registry_state +{ + stable_context_vec_t registry = + stable_context_vec_t{reserve_size_t{stable_context_vec_t::chunk_size}}; + published_snapshot_t published = {}; +}; + +registry_state* +get_registry_state() +{ + static auto*& _v = common::static_object::construct(); + return _v; +} + +stable_context_vec_t* +get_registered_contexts_impl() +{ + auto* _state = get_registry_state(); + return (_state) ? &_state->registry : nullptr; +} + +published_snapshot_t* +get_published_snapshot() +{ + auto* _state = get_registry_state(); + return (_state) ? &_state->published : nullptr; +} + +context_snapshot_ptr_t +current_registered_contexts() +{ + auto* _pub = get_published_snapshot(); + if(!_pub) return {}; + return load_snapshot(*_pub); +} + +// Rebuilds the published snapshot from the registry. Must be called with get_contexts_mutex() +// held, i.e. by whichever function just mutated the registry. +void +publish_registered_contexts() +{ + auto* _pub = get_published_snapshot(); + if(!_pub) return; + + auto _data = std::make_shared(); + + if(auto* _impl = get_registered_contexts_impl()) + { + _data->owned.reserve(_impl->size()); + _data->view.reserve(_impl->size()); + for(const auto& itr : *_impl) + { + if(!itr) continue; + _data->owned.emplace_back(itr); + _data->view.emplace_back(itr.get()); + } + } + + store_snapshot(*_pub, std::move(_data)); +} + +// The registry is index-addressed: allocate_context() derives context_idx from the slot position, +// so a context id maps straight back to its slot. +context_ptr_t +get_registered_context_slot(uint64_t handle) +{ + auto* _impl = get_registered_contexts_impl(); + if(!_impl || handle < get_contexts_offset()) return {}; + + auto _idx = handle - get_contexts_offset(); + if(_idx >= _impl->size()) return {}; + + return _impl->at(_idx); +} + +// Context ids whose stop is in progress. stop_context() releases get_contexts_mutex() across the +// GPU drain, so for that window the context is still in the active array with no lock held. This +// set is what keeps start_context() and a second stop_context() from acting on that half-stopped +// state. Guarded by get_contexts_mutex(). +std::unordered_set& +get_stopping_contexts() +{ + static auto _v = std::unordered_set{}; + return _v; +} + +std::condition_variable& +get_contexts_cv() +{ + static auto _v = std::condition_variable{}; + return _v; +} + } // namespace +// Waits until no context is mid-stop, which restores the property the old single-lock +// stop_context() had: no other lifecycle operation ever observes a context whose services have +// been torn down but whose active slot is still populated. +void +wait_for_stopping_contexts(std::unique_lock& _lk) +{ + get_contexts_cv().wait(_lk, []() { return get_stopping_contexts().empty(); }); +} + +std::mutex& +get_contexts_mutex() +{ + static auto _v = std::mutex{}; + return _v; +} + +bool +dispatch_counter_collection_service::intersects( + const dispatch_counter_collection_service& rhs) const +{ + if(agents.empty() || rhs.agents.empty()) return true; + const auto& small = (agents.size() < rhs.agents.size()) ? agents : rhs.agents; + const auto& large = (agents.size() < rhs.agents.size()) ? rhs.agents : agents; + for(const auto& agent_id : small) + { + if(large.count(agent_id) > 0) return true; + } + return false; +} + +registered_contexts_snapshot::registered_contexts_snapshot(context_snapshot_ptr_t&& data) +: m_data{std::move(data)} +{ + if(m_data) + { + m_begin = m_data->view.data(); + m_end = m_begin + m_data->view.size(); + } +} + +registered_contexts_snapshot +get_registered_contexts_snapshot() +{ + return registered_contexts_snapshot{current_registered_contexts()}; +} + context_array_t& get_registered_contexts(context_array_t& data, context_filter_t filter) { data.clear(); - if(!get_registered_contexts_impl()) return data; - - auto num_ctx = get_registered_contexts_impl()->size(); - if(num_ctx <= 0) return data; + auto snapshot = get_registered_contexts_snapshot(); + if(snapshot.empty()) return data; - data.reserve(num_ctx); - for(auto& itr : *get_registered_contexts_impl()) + data.reserve(snapshot.size()); + for(const auto* ctx : snapshot) { - const auto* ctx = &itr.value(); - if(ctx) - { - if(!filter || (filter && filter(ctx))) data.emplace_back(ctx); - } + if(!filter || (filter && filter(ctx))) data.emplace_back(ctx); } return data; } @@ -219,13 +393,9 @@ allocate_context() // initial context identifier number auto _idx = get_registered_contexts_impl()->size() + get_contexts_offset(); - // make space in registered - get_registered_contexts_impl()->emplace_back(); - // create an entry in the registered - auto& _cfg_v = get_registered_contexts_impl()->back(); - _cfg_v.emplace(); - auto* _cfg = &_cfg_v.value(); + auto& _cfg_v = get_registered_contexts_impl()->emplace_back(std::make_shared()); + auto* _cfg = _cfg_v.get(); // ... if(!_cfg) return std::nullopt; @@ -238,20 +408,15 @@ allocate_context() << " rocprofiler internal error: a context was allocated without an associated tool client " "identifier"; + publish_registered_contexts(); + return rocprofiler_context_id_t{_idx}; } context* get_mutable_registered_context(rocprofiler_context_id_t id) { - if(id.handle < get_contexts_offset()) return nullptr; - if(!get_registered_contexts_impl()) return nullptr; - auto _idx = id.handle - get_contexts_offset(); - if(_idx >= get_registered_contexts_impl()->size()) - return nullptr; - else if(get_registered_contexts_impl()->at(_idx).has_value()) - return &get_registered_contexts_impl()->at(_idx).value(); - return nullptr; + return get_registered_context_slot(id.handle).get(); } const context* @@ -271,38 +436,52 @@ start_context(rocprofiler_context_id_t context_id) { if(context_id.handle < get_contexts_offset()) return ROCPROFILER_STATUS_ERROR_CONTEXT_NOT_FOUND; - if(!get_registered_contexts_impl()) return ROCPROFILER_STATUS_ERROR_CONTEXT_NOT_FOUND; - - if((context_id.handle - get_contexts_offset()) >= get_registered_contexts_impl()->size()) - return ROCPROFILER_STATUS_ERROR_CONTEXT_NOT_FOUND; - - const auto* cfg = get_registered_context(context_id); + // Held for the rest of the function: the service starts below run without the contexts mutex, + // and deregister_client_contexts() may drop the registry's reference while they do. + auto _owner = context_ptr_t{}; + const context* cfg = nullptr; + { + auto _lk = std::unique_lock{get_contexts_mutex()}; + _owner = get_registered_context_slot(context_id.handle); + cfg = _owner.get(); + } if(!cfg) return ROCPROFILER_STATUS_ERROR_CONTEXT_NOT_FOUND; if(validate_context(cfg) != ROCPROFILER_STATUS_SUCCESS) return ROCPROFILER_STATUS_ERROR_CONTEXT_INVALID; - auto current_contexts = context_array_t{}; - for(const auto* itr : get_active_contexts(current_contexts)) - { - if(cfg->context_idx == itr->context_idx) - { - return ROCPROFILER_STATUS_SUCCESS; - } - else if(cfg->dispatch_counter_collection && itr->dispatch_counter_collection) - { - // conflicting context - return ROCPROFILER_STATUS_ERROR_CONTEXT_CONFLICT; - } - } - uint64_t rocp_tot_contexts = get_registered_contexts_impl()->size(); auto idx = rocp_tot_contexts; auto& active_contexts = get_active_contexts_impl(); { // hold a lock here to prevent multiple threads from finding the same nullptr slot auto _lk = std::unique_lock{get_contexts_mutex()}; + + // A context that is mid-stop is still in the active array while its GPU drain runs, so + // scanning against that state would either report a conflict against a context that is on + // its way out, or hand back a slot that the stop is about to clear. + wait_for_stopping_contexts(_lk); + + auto current_contexts = context_array_t{}; + for(const auto* itr : get_active_contexts(current_contexts)) + { + if(cfg->context_idx == itr->context_idx) + { + return ROCPROFILER_STATUS_SUCCESS; + } + else if(cfg->dispatch_counter_collection && itr->dispatch_counter_collection && + cfg->dispatch_counter_collection->intersects(*itr->dispatch_counter_collection)) + { + // Conflicting context. Two counter-collection contexts can run concurrently as + // long as they target disjoint sets of GPU agents -- the hardware counters they + // program are per-agent, so contexts that never touch the same agent cannot + // contend. A context with no agent restriction claims every agent and therefore + // still conflicts with any other counter-collection context. + return ROCPROFILER_STATUS_ERROR_CONTEXT_CONFLICT; + } + } + // try to find a nullptr slot first for(size_t i = 0; i < active_contexts.size(); ++i) { @@ -317,6 +496,7 @@ start_context(rocprofiler_context_id_t context_id) return ROCPROFILER_STATUS_SUCCESS; } } + // if no nullptr slot was found, then create one while lock is held if(idx == rocp_tot_contexts) { @@ -331,8 +511,7 @@ start_context(rocprofiler_context_id_t context_id) // atomic swap the pointer into the "active" array used internally const context* _expected = nullptr; - bool success = active_contexts.at(idx).compare_exchange_strong( - _expected, get_registered_context(context_id)); + bool success = active_contexts.at(idx).compare_exchange_strong(_expected, cfg); if(!success) { @@ -366,65 +545,116 @@ start_context(rocprofiler_context_id_t context_id) rocprofiler_status_t stop_context(rocprofiler_context_id_t idx) { - // hold a lock here to prevent other thread from changing the active contexts array - auto _lk = std::unique_lock{get_contexts_mutex()}; - - // atomically assign the context pointer to NULL so that it is skipped in future - // callbacks - for(auto& itr : get_active_contexts_impl()) + std::atomic* slot = nullptr; + const context* _expected = nullptr; + // Held for the rest of the function. The teardown below runs without the contexts mutex, so + // without this the registry's reference is the only thing keeping the context alive and + // deregister_client_contexts() could destroy it out from under the drain. + auto _owner = context_ptr_t{}; + + // Phase one, locked: claim the context. Only the search of the active array needs the lock; + // the teardown below must not hold it. { - const context* _expected = itr.load(std::memory_order_acquire); - if(_expected && _expected->context_idx == idx.handle) - { - bool success = itr.compare_exchange_strong(_expected, nullptr); + auto _lk = std::unique_lock{get_contexts_mutex()}; - if(success) + // another thread may already be tearing this context down + wait_for_stopping_contexts(_lk); + + for(auto& itr : get_active_contexts_impl()) + { + const context* _ctx = itr.load(std::memory_order_acquire); + if(_ctx && _ctx->context_idx == idx.handle) { - auto nactive = get_num_active_contexts().load(std::memory_order_acquire); - if(nactive > 0) get_num_active_contexts().fetch_sub(1, std::memory_order_release); + slot = &itr; + _expected = _ctx; + break; + } + } - if(_expected->dispatch_counter_collection) - { - rocprofiler::counters::stop_context(const_cast(_expected)); - } + if(!slot) return ROCPROFILER_STATUS_ERROR_CONTEXT_NOT_FOUND; - if(_expected->dispatch_spm) - rocprofiler::spm::stop_context(const_cast(_expected)); + _owner = get_registered_context_slot(idx.handle); + get_stopping_contexts().emplace(idx.handle); + } - if(_expected->device_thread_trace) _expected->device_thread_trace->stop_context(); - if(_expected->dispatch_thread_trace) - _expected->dispatch_thread_trace->stop_context(); + auto _unclaim = common::scope_destructor{[&idx]() { + { + auto _lk = std::unique_lock{get_contexts_mutex()}; + get_stopping_contexts().erase(idx.handle); + } + get_contexts_cv().notify_all(); + }}; + + // Phase two, unlocked: the service teardowns below call hsa::queue_controller_sync(), an + // unbounded wait on in-flight GPU work. Holding get_contexts_mutex() across it stalls every + // context lifecycle operation in the process behind one context's dispatches, and it puts the + // mutex on the far side of a wait that the completion path has to get through -- so any future + // completion-path read that took the mutex would deadlock rather than merely block. + // + // The active slot stays populated for the whole phase, which is deliberate: + // kernel_dispatch_phase_enter_hook has to keep seeing the context so the serialized -> + // unserialized transition stays coordinated until disable_serialization() has run. Clearing + // the slot first opens the opposite window, in which the enter hook sees no active context and + // a dispatch is submitted without serializer packets while the serializer is still enabled. + // get_stopping_contexts() is what keeps other lifecycle callers from reading that state as a + // running context. + if(_expected->dispatch_counter_collection) + { + rocprofiler::counters::stop_context(const_cast(_expected)); + } - rocprofiler::hsa::queue_interposition:: - notify_queue_interposition_consumer_context_stopped(_expected); + if(_expected->dispatch_spm) rocprofiler::spm::stop_context(const_cast(_expected)); - if(_expected->device_counter_collection) - { - rocprofiler::counters::stop_agent_ctx(const_cast(_expected)); - } + if(_expected->device_thread_trace) _expected->device_thread_trace->stop_context(); + if(_expected->dispatch_thread_trace) _expected->dispatch_thread_trace->stop_context(); -#if ROCPROFILER_SDK_HSA_PC_SAMPLING > 0 - if(_expected->pc_sampler) - { - rocprofiler::pc_sampling::stop_service(_expected); - } -#endif + // Phase three, relocked: retire the slot. The element address is stable across phase two -- + // stable_vector never moves an element once constructed -- so slot is still the entry the + // search found. + { + auto _lk = std::unique_lock{get_contexts_mutex()}; - return ROCPROFILER_STATUS_SUCCESS; - } + if(!slot->compare_exchange_strong(_expected, nullptr)) + { + // Not reachable as written: deactivate_client_contexts() is the only other writer of + // this slot and the stopping marker holds it off until phase four returns. Kept + // because the pre-phasing code bailed out the same way on a lost exchange. + return ROCPROFILER_STATUS_ERROR_CONTEXT_NOT_FOUND; } + + auto nactive = get_num_active_contexts().load(std::memory_order_acquire); + if(nactive > 0) get_num_active_contexts().fetch_sub(1, std::memory_order_release); + } + + // Phase four, unlocked: the remaining services are not queue-interposed, so they only need + // the slot to be cleared first, not the lock to be held. They wait on the GPU too -- + // stop_agent_ctx() waits for its stop packet and pc_sampling::stop_service() drains the + // device buffers -- and the stopping marker above, not the mutex, is what keeps a concurrent + // start_context() from running before they finish. + rocprofiler::hsa::queue_interposition::notify_queue_interposition_consumer_context_stopped( + _expected); + + if(_expected->device_counter_collection) + { + rocprofiler::counters::stop_agent_ctx(const_cast(_expected)); } - return ROCPROFILER_STATUS_ERROR_CONTEXT_NOT_FOUND; // compare exchange failed +#if ROCPROFILER_SDK_HSA_PC_SAMPLING > 0 + if(_expected->pc_sampler) + { + rocprofiler::pc_sampling::stop_service(_expected); + } +#endif + + return ROCPROFILER_STATUS_SUCCESS; } context_id_array_t get_client_contexts(rocprofiler_client_id_t id) { auto _data = context_id_array_t{}; - if(!get_registered_contexts_impl()) return _data; - for(auto& itr : *get_registered_contexts_impl()) + for(const auto* itr : get_registered_contexts_snapshot()) { if(itr->client_idx == id.handle) { @@ -440,7 +670,11 @@ stop_client_contexts(rocprofiler_client_id_t client_id) if(!get_registered_contexts_impl()) return ROCPROFILER_STATUS_ERROR; auto ret = ROCPROFILER_STATUS_SUCCESS; - for(auto& itr : *get_registered_contexts_impl()) + + // Held across the loop: stop_context() may block on a GPU drain, and a snapshot is what keeps + // the contexts it is iterating alive for that long. + auto snapshot = get_registered_contexts_snapshot(); + for(const auto* itr : snapshot) { if(itr->client_idx == client_id.handle) { @@ -457,6 +691,11 @@ void deactivate_client_contexts(rocprofiler_client_id_t client_id) { auto _lk = std::unique_lock{get_contexts_mutex()}; + + // a context mid-stop is still in the active array; let its teardown finish rather than + // clearing the slot underneath it + wait_for_stopping_contexts(_lk); + for(auto& itr : get_active_contexts_impl()) { const context* itr_v = itr.load(std::memory_order_acquire); @@ -474,16 +713,38 @@ deactivate_client_contexts(rocprofiler_client_id_t client_id) void deregister_client_contexts(rocprofiler_client_id_t client_id) { - for(auto& itr : *get_registered_contexts_impl()) + if(!get_registered_contexts_impl()) return; + + // Outlives the lock below so the contexts are destroyed without it held: a context destructor + // calls back into HSA, which is not something to do underneath the registry mutex. + auto _retired = std::vector{}; + { - if(itr->client_idx == client_id.handle && buffer::get_buffers()) + // Mutates the registry, so it needs the same lock allocate_context() takes. + auto _lk = std::unique_lock{get_contexts_mutex()}; + + // a context mid-stop is still being torn down through a raw pointer; retiring its registry + // entry now would pull the object out from under that teardown + wait_for_stopping_contexts(_lk); + + for(auto& itr : *get_registered_contexts_impl()) { - for(auto& bitr : *buffer::get_buffers()) + if(!itr) continue; + + if(itr->client_idx == client_id.handle && buffer::get_buffers()) { - if(bitr && bitr->context_id == itr->context_idx) bitr.reset(); + for(auto& bitr : *buffer::get_buffers()) + { + if(bitr && bitr->context_id == itr->context_idx) bitr.reset(); + } + // Moved rather than reset: dropping the registry's reference here would destroy + // the context under the lock, and a reader holding an older snapshot still needs + // it to stay alive until that snapshot is released. + _retired.emplace_back(std::move(itr)); } - itr.reset(); } + + publish_registered_contexts(); } } diff --git a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/context/context.hpp b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/context/context.hpp index 73f075f9b148..c1c4c376c2ad 100644 --- a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/context/context.hpp +++ b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/context/context.hpp @@ -35,13 +35,17 @@ #include #include +#include +#include #include #include #include #include +#include #include #include +#include namespace rocprofiler { @@ -82,10 +86,27 @@ struct dispatch_counter_collection_service // Contains callback information along with other data needed to collect/process // counters. std::vector> callbacks{}; + // GPU agents this context collects on. An empty set means every GPU agent, which is what a + // plain configure_{buffer,callback}_dispatch call produces; + // rocprofiler_dispatch_counting_service_set_agents narrows it. The set is read on the + // dispatch path and when scoping serialization, so it is fixed at configure time and never + // mutated once the context is started. + std::unordered_set agents{}; // A flag to state whether 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). common::Synchronized enabled{false}; + + // An empty agent set means "all GPU agents", so an unrestricted context keeps behaving + // exactly as it did before per-agent scoping existed. + bool collects_on(rocprofiler_agent_id_t agent_id) const + { + return agents.empty() || agents.count(agent_id) > 0; + } + + // Two counter-collection contexts can coexist as long as they do not both want the same + // GPU. An unrestricted context claims every agent, so it intersects with everything. + bool intersects(const dispatch_counter_collection_service& rhs) const; }; struct spm_dispatch_counter_collection_service @@ -196,6 +217,40 @@ stop_context(rocprofiler_context_id_t id); using context_array_t = common::container::small_vector; using context_id_array_t = common::container::small_vector; +// defined in context.cpp +struct registered_context_storage; + +/// \brief An immutable view of the registered contexts, taken at the point of construction. +/// +/// The registry itself may only be walked with get_contexts_mutex() held: allocate_context() +/// appends to it while dispatch and completion paths read it. The completion path cannot take +/// that mutex, because stop_context() waits on the GPU drain and the drain waits on completions, +/// so it reads a snapshot instead. A snapshot also owns the contexts it names, which is what makes +/// it safe against deregister_client_contexts() destroying a context mid-iteration. +class registered_contexts_snapshot +{ +public: + registered_contexts_snapshot() = default; + + const context* const* begin() const { return m_begin; } + const context* const* end() const { return m_end; } + size_t size() const { return m_end - m_begin; } + bool empty() const { return m_begin == m_end; } + +private: + friend registered_contexts_snapshot get_registered_contexts_snapshot(); + + explicit registered_contexts_snapshot(std::shared_ptr&&); + + std::shared_ptr m_data = {}; + const context* const* m_begin = nullptr; + const context* const* m_end = nullptr; +}; + +/// \brief Returns the currently published snapshot of the registered contexts. +registered_contexts_snapshot +get_registered_contexts_snapshot(); + context* get_mutable_registered_context(rocprofiler_context_id_t id); @@ -207,6 +262,10 @@ using context_filter_t = bool (*)(const context*); inline bool default_context_filter(const context* val); +// These read the published snapshot and then drop it, so the pointers they return outlive the +// guarantee that the contexts are alive. That is fine for callers that cannot overlap with +// deregister_client_contexts(); anything on a dispatch or completion path should hold a +// registered_contexts_snapshot for as long as it uses the pointers. context_array_t& get_registered_contexts(context_array_t& data, context_filter_t filter = default_context_filter); @@ -242,6 +301,20 @@ deactivate_client_contexts(rocprofiler_client_id_t id); void deregister_client_contexts(rocprofiler_client_id_t id); +// The mutex that serialises start_context / stop_context and any mutation that must be +// atomic with activation state (e.g. setting the agent restriction on a context). +std::mutex& +get_contexts_mutex(); + +// Blocks until no context stop is in progress. stop_context() releases get_contexts_mutex() +// across the GPU drain, so holding the mutex alone is not enough to exclude a stop: for that +// window the context is still in the active array with nothing held. Anything that reads +// activation state to decide whether a context is running has to wait here first. +// +// _lk must own get_contexts_mutex(); it is released while waiting and re-acquired on return. +void +wait_for_stopping_contexts(std::unique_lock& _lk); + inline bool default_context_filter(const context* val) { diff --git a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/CMakeLists.txt b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/CMakeLists.txt index 0f3b95f9dd55..7d3e09a55c0b 100644 --- a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/CMakeLists.txt +++ b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/CMakeLists.txt @@ -9,7 +9,8 @@ set(ROCPROFILER_LIB_COUNTERS_SOURCES controller.cpp device_counting.cpp ioctl.cpp - firmware_restrictions.cpp) + firmware_restrictions.cpp + queue_hooks.cpp) set(ROCPROFILER_LIB_COUNTERS_HEADERS metrics.hpp dimensions.hpp @@ -22,7 +23,8 @@ set(ROCPROFILER_LIB_COUNTERS_HEADERS device_counting.hpp sample_consumer.hpp ioctl.hpp - firmware_restrictions.hpp) + firmware_restrictions.hpp + queue_hooks.hpp) target_sources(rocprofiler-sdk-object-library PRIVATE ${ROCPROFILER_LIB_COUNTERS_SOURCES} ${ROCPROFILER_LIB_COUNTERS_HEADERS}) add_subdirectory(parser) diff --git a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/core.cpp b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/core.cpp index f53406a9104c..576752a2ab46 100644 --- a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/core.cpp +++ b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/core.cpp @@ -25,6 +25,7 @@ #include "lib/common/container/small_vector.hpp" #include "lib/common/synchronized.hpp" #include "lib/common/utility.hpp" +#include "lib/rocprofiler-sdk/agent.hpp" #include "lib/rocprofiler-sdk/aql/packet_construct.hpp" #include "lib/rocprofiler-sdk/context/context.hpp" #include "lib/rocprofiler-sdk/counters/dispatch_handlers.hpp" @@ -152,7 +153,9 @@ start_context(const context::context* ctx) auto* controller = hsa::get_queue_controller(); bool already_enabled = true; - CHECK_NOTNULL(controller)->enable_serialization(); + // Scope serialization to the agents this context collects on. An empty set still means + // every agent, so an unrestricted context serializes the whole machine as before. + CHECK_NOTNULL(controller)->enable_serialization(ctx->dispatch_counter_collection->agents); ctx->dispatch_counter_collection->enabled.wlock([&](auto& enabled) { if(enabled) return; already_enabled = false; @@ -161,48 +164,10 @@ start_context(const context::context* ctx) if(!already_enabled) { + // Counter collection no longer registers a per-queue callback with the queue + // controller; the HSA write interceptor calls counters::kernel_dispatch_phase_enter_hook / + // kernel_dispatch_phase_exit_hook directly (see hsa/queue.cpp). Keep the callback thread. callback_thread_start(); - - for(auto& cb : ctx->dispatch_counter_collection->callbacks) - { - using external_corr_id_map_t = tracing::external_correlation_id_map_t; - - // 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, - hsa::queue_callbacks_t{.batch_packets = []() { return false; }, - .write_interceptor = - [=](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 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 - .signal_completion = - [=](const hsa::Queue& /*queue*/, - hsa::rocprofiler_packet /*kern_pkt*/, - std::shared_ptr& session, - hsa::packet_data_t& packet, - inst_pkt_t& aql, - kernel_dispatch::profiling_time dispatch_time) { - completed_cb( - ctx, cb, session, packet, aql, dispatch_time); - }}); - } } } @@ -220,28 +185,74 @@ stop_context(const context::context* ctx) if(controller) { - controller->disable_serialization(); - - // In attach mode, remove counter callbacks when stopping the context so - // new dispatches during detach drain can pass through without incrementing - // _active_kernels. In normal profiling, keep callbacks registered to avoid - // dropping counter data for in-flight dispatches. - if(registration::is_attached()) - { - for(auto& cb : ctx->dispatch_counter_collection->callbacks) - { - if(cb->queue_id != rocprofiler::hsa::ClientID{-1}) - { - controller->remove_callback(cb->queue_id); - cb->queue_id = rocprofiler::hsa::ClientID{-1}; - } - } - } + // Drain in-flight dispatches before anything else is torn down. The review of #8891 + // accepted provenance-based completion routing on the condition that the callback thread + // and the counter_callback_info objects stay alive until in-flight dispatches drain, so the + // drain is what makes that condition hold literally rather than by argument. + // + // context::stop_context calls this function while the context is still in the active list, + // which is what keeps the service visible for the duration of the drain: the enter hook + // still reaches queue_cb, and queue_cb's disabled path returns serialize=true so the + // serialized->unserialized transition stays coordinated until disable_serialization() runs + // below. That ordering is why no separate "draining" flag is needed -- but it only works + // because the drain happens here, before the slot is cleared. + hsa::queue_controller_sync(); + controller->disable_serialization(ctx->dispatch_counter_collection->agents); + // No per-queue callback to remove; counters::kernel_dispatch_phase_enter_hook no-ops once + // dispatch_counter_collection is disabled above. } + // After the drain. consumer_thread_t::exit() also waits until its queue is empty before + // joining, and add() consumes inline once the thread is gone, so a late completion is still + // processed -- but that is a backstop, not the guarantee the review asked for. callback_thread_stop(); } +rocprofiler_status_t +set_dispatch_agents(rocprofiler_context_id_t context_id, + const rocprofiler_agent_id_t* agents, + size_t num_agents) +{ + if(num_agents > 0 && agents == nullptr) return ROCPROFILER_STATUS_ERROR_INVALID_ARGUMENT; + + auto* ctx_p = context::get_mutable_registered_context(context_id); + if(!ctx_p) return ROCPROFILER_STATUS_ERROR_CONTEXT_INVALID; + if(!ctx_p->dispatch_counter_collection) return ROCPROFILER_STATUS_ERROR_CONTEXT_NOT_FOUND; + + // The agent set is read without a lock on the dispatch path and is what scopes + // serialization at start, so it may only change while the context is stopped. Hold the + // contexts mutex so this check and the assignment are atomic with start_context: a + // concurrent start could otherwise activate the context after the scan (using the old + // set for serialization) and then race with the unordered-set assignment below while + // dispatch hooks read it. + auto _lk = std::unique_lock{context::get_contexts_mutex()}; + + // A context that is mid-stop is still in the active array while its GPU drain runs, and that + // drain happens with the mutex released, so the lock alone does not exclude it. Without this + // the scan below would read a context that is on its way out as a running one. + context::wait_for_stopping_contexts(_lk); + + for(const auto* itr : context::get_active_contexts()) + { + if(itr && itr->context_idx == ctx_p->context_idx) + return ROCPROFILER_STATUS_ERROR_CONFIGURATION_LOCKED; + } + + auto selected = std::unordered_set{}; + selected.reserve(num_agents); + for(size_t i = 0; i < num_agents; ++i) + { + const auto* agent = rocprofiler::agent::get_agent(agents[i]); + if(!agent || agent->type != ROCPROFILER_AGENT_TYPE_GPU) + return ROCPROFILER_STATUS_ERROR_AGENT_NOT_FOUND; + selected.emplace(agents[i]); + } + + ctx_p->dispatch_counter_collection->agents = std::move(selected); + + return ROCPROFILER_STATUS_SUCCESS; +} + rocprofiler_status_t configure_agent_collection(rocprofiler_context_id_t context_id, rocprofiler_buffer_id_t buffer_id, diff --git a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/core.hpp b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/core.hpp index 588098cedad5..07a6a3a7b22f 100644 --- a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/core.hpp +++ b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/core.hpp @@ -52,9 +52,6 @@ struct counter_callback_info rocprofiler_context_id_t context{.handle = 0}; // Link to the internal context this is associated with const context::context* internal_context{nullptr}; - // HSA Queue ClientID. This is an ID we get when we insert a callback into the - // HSA queue interceptor. This ID can be used to disable the callback. - rocprofiler::hsa::ClientID queue_id{-1}; // Buffer to use for storing counter data. Used if callback is not set. std::optional buffer; @@ -91,6 +88,11 @@ configure_callback_dispatch(rocprofiler_context_id_t context_i rocprofiler_dispatch_counting_record_cb_t record_callback, void* record_callback_args); +rocprofiler_status_t +set_dispatch_agents(rocprofiler_context_id_t context_id, + const rocprofiler_agent_id_t* agents, + size_t num_agents); + rocprofiler_status_t configure_agent_collection(rocprofiler_context_id_t context_id, rocprofiler_buffer_id_t buffer_id, diff --git a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/device_counting.cpp b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/device_counting.cpp index 22ed73bbaae0..7c8755c1a23c 100644 --- a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/device_counting.cpp +++ b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/device_counting.cpp @@ -698,7 +698,17 @@ device_counting_service_hsa_registration() agent_callback_data::~agent_callback_data() { - if(completion.handle != 0) hsa::get_core_table()->hsa_signal_destroy_fn(completion); + if(completion.handle == 0) return; + + // Guarded the same way as hsa::active_signals::destroy(). The owning context is not + // guaranteed to be destroyed while HSA is still up -- teardown can reach here after the + // runtime has been shut down, and the table pointer is also null under unit test -- so + // reaching through the table unconditionally faults instead of leaking a signal. + const auto* _core_table = hsa::get_core_table(); + if(hsa::get_hsa_ref_count() > 0 && _core_table && _core_table->hsa_signal_destroy_fn) + { + _core_table->hsa_signal_destroy_fn(completion); + } } } // namespace counters } // namespace rocprofiler diff --git a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/queue_hooks.cpp b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/queue_hooks.cpp new file mode 100644 index 000000000000..e7464db13e3d --- /dev/null +++ b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/queue_hooks.cpp @@ -0,0 +1,134 @@ +// MIT License +// +// Copyright (c) 2026 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/counters/queue_hooks.hpp" +#include "lib/rocprofiler-sdk/context/context.hpp" +#include "lib/rocprofiler-sdk/counters/dispatch_handlers.hpp" +#include "lib/rocprofiler-sdk/hsa/agent_cache.hpp" +#include "lib/rocprofiler-sdk/hsa/queue.hpp" +#include "lib/rocprofiler-sdk/hsa/queue_hooks/client_ids.hpp" + +namespace rocprofiler +{ +namespace counters +{ +namespace +{ +auto +counter_contexts_filter() +{ + return [](const context::context* ctx) -> bool { + return ctx && ctx->dispatch_counter_collection != nullptr; + }; +} +} // namespace + +void +kernel_dispatch_phase_enter_hook( + const hsa::Queue& queue, + const hsa::rocprofiler_packet& kernel_packet, + rocprofiler_kernel_id_t kernel_id, + rocprofiler_dispatch_id_t dispatch_id, + rocprofiler_user_data_t* user_data, + const hsa::queue_info_session_t::external_corr_id_map_t& ext_corr_ids, + const context::correlation_id* correlation_id, + hsa::inst_pkt_t& inst_pkt, + bool& is_serialized) +{ + const auto agent_id = CHECK_NOTNULL(queue.get_agent().get_rocp_agent())->id; + + auto active = context::get_active_contexts(counter_contexts_filter()); + for(const auto* ctx : active) + { + // Skip contexts that do not collect on this dispatch's agent. This is what keeps a + // GPU-1-only context from serializing GPU-0: queue_cb is what returns serialize=true, + // including on its disabled path, so filtering has to happen before the call rather + // than inside it. + if(!ctx->dispatch_counter_collection->collects_on(agent_id)) continue; + + for(auto& cb : ctx->dispatch_counter_collection->callbacks) + { + auto [packet, bSerial] = queue_cb(ctx, + cb, + queue, + kernel_packet, + kernel_id, + dispatch_id, + user_data, + ext_corr_ids, + correlation_id); + if(packet) + inst_pkt.emplace_back(std::move(packet), hsa::queue_hooks::COUNTERS_CLIENT_ID); + is_serialized |= bSerial; + } + } +} + +void +kernel_dispatch_phase_exit_hook(const hsa::Queue* /*queue*/, + const hsa::rocprofiler_packet& /*kernel_packet*/, + std::shared_ptr& session, + hsa::packet_data_t& packet, + hsa::inst_pkt_t& inst_pkt, + kernel_dispatch::profiling_time dispatch_time) +{ + bool has_counter_packets = false; + for(const auto& tagged_pkt : inst_pkt) + { + if(tagged_pkt.second == hsa::queue_hooks::COUNTERS_CLIENT_ID) + { + has_counter_packets = true; + break; + } + } + if(!has_counter_packets) return; + + // Route by packet provenance, not current activeness: completed_cb self-filters via + // packet_return_map, so in-flight dispatches still complete after stop_context removes the + // context from the active list. This is what guarantees a completion that arrives is delivered; + // the drain in stop_context is what bounds when completions arrive. Kernel replay needs the + // same property, since each pass completes separately. + // + // Read through a snapshot rather than the registry itself. This runs on the HSA async signal + // handler, so it cannot take get_contexts_mutex() -- stop_context() waits on the drain that + // this completion is part of -- and the registry is not safe to walk unlocked while + // allocate_context() appends to it. Holding the snapshot also keeps every context it names + // alive for the duration of the loop. + auto contexts = context::get_registered_contexts_snapshot(); + for(const auto* ctx : contexts) + { + if(!counter_contexts_filter()(ctx)) continue; + + for(auto& cb : ctx->dispatch_counter_collection->callbacks) + { + completed_cb(ctx, cb, session, packet, inst_pkt, dispatch_time); + } + } +} + +bool +is_any_active() +{ + return !context::get_active_contexts(counter_contexts_filter()).empty(); +} +} // namespace counters +} // namespace rocprofiler diff --git a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/queue_hooks.hpp b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/queue_hooks.hpp new file mode 100644 index 000000000000..1a76e7d66b3e --- /dev/null +++ b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/queue_hooks.hpp @@ -0,0 +1,72 @@ +// MIT License +// +// Copyright (c) 2026 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 "lib/rocprofiler-sdk/context/context.hpp" +#include "lib/rocprofiler-sdk/hsa/queue.hpp" +#include "lib/rocprofiler-sdk/hsa/queue_info_session.hpp" +#include "lib/rocprofiler-sdk/hsa/rocprofiler_packet.hpp" +#include "lib/rocprofiler-sdk/kernel_dispatch/profiling_time.hpp" + +#include + +namespace rocprofiler +{ +namespace counters +{ +// Explicit replacement for the dispatch-counter per-queue callback that used to +// be registered with the HSA queue controller. Iterates active +// dispatch_counter_collection contexts and calls each callback's queue_cb; +// appends produced packets to inst_pkt, OR-folding each callback's serialize flag +// into is_serialized. +void +kernel_dispatch_phase_enter_hook( + const hsa::Queue& queue, + const hsa::rocprofiler_packet& kernel_packet, + rocprofiler_kernel_id_t kernel_id, + rocprofiler_dispatch_id_t dispatch_id, + rocprofiler_user_data_t* user_data, + const hsa::queue_info_session_t::external_corr_id_map_t& ext_corr_ids, + const context::correlation_id* correlation_id, + hsa::inst_pkt_t& inst_pkt, + bool& is_serialized); + +// Explicit replacement for the dispatch-counter completion callback. Iterates +// registered dispatch_counter_collection contexts (not only active ones) and calls +// each callback's completed_cb; completed_cb self-filters via packet_return_map so +// in-flight dispatches still complete after stop_context. +// queue is nullable: the hook never dereferences it, and tests exercise the +// early-return path without an HSA runtime to build a queue from. +void +kernel_dispatch_phase_exit_hook(const hsa::Queue* queue, + const hsa::rocprofiler_packet& kernel_packet, + std::shared_ptr& session, + hsa::packet_data_t& packet, + hsa::inst_pkt_t& inst_pkt, + kernel_dispatch::profiling_time dispatch_time); + +// True if any context currently has dispatch counter collection active. +bool +is_any_active(); +} // namespace counters +} // namespace rocprofiler diff --git a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/sample_processing.cpp b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/sample_processing.cpp index a6ca2609f070..3dc277df221d 100644 --- a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/sample_processing.cpp +++ b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/sample_processing.cpp @@ -167,18 +167,26 @@ callback_thread_get() static auto*& _v = common::static_object::construct(&process_completed_cb); return *CHECK_NOTNULL(_v); } + +// Reference count for the callback consumer thread. Incremented by callback_thread_start() +// for each context that starts and decremented by callback_thread_stop() when each stops; +// the thread is only joined when the count returns to zero. Two dispatch-counting contexts +// with disjoint agent sets may be active at the same time, so without the refcount stopping +// either one would kill the consumer while the other still needs it. +std::atomic callback_thread_refcount{0}; } // namespace void callback_thread_start() { + ++callback_thread_refcount; callback_thread_get().start(); } void callback_thread_stop() { - callback_thread_get().exit(); + if(--callback_thread_refcount == 0) callback_thread_get().exit(); } /** diff --git a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/tests/CMakeLists.txt b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/tests/CMakeLists.txt index 14d183af7d09..b830ef7908b8 100644 --- a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/tests/CMakeLists.txt +++ b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/tests/CMakeLists.txt @@ -61,8 +61,16 @@ target_link_libraries( rocprofiler-sdk::rocprofiler-sdk-hsa-runtime) set(ROCPROFILER_LIB_COUNTER_TEST_SOURCES - metrics_test.cpp evaluate_ast_test.cpp dimension.cpp core.cpp code_object_loader.cpp - device_counting.cpp firmware_restrictions_test.cpp local_context.cpp) + metrics_test.cpp + evaluate_ast_test.cpp + dimension.cpp + core.cpp + code_object_loader.cpp + device_counting.cpp + firmware_restrictions_test.cpp + queue_hooks_test.cpp + per_agent_scoping_test.cpp + local_context.cpp) set(ROCPROFILER_LIB_COUNTER_TEST_HEADERS code_object_loader.hpp device_counting.hpp) add_executable(counter-tests) diff --git a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/tests/core.cpp b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/tests/core.cpp index ac3c77f663ee..26e77c265324 100644 --- a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/tests/core.cpp +++ b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/tests/core.cpp @@ -560,19 +560,12 @@ TEST(core, start_stop_buffered_ctx) ASSERT_TRUE(ctx.dispatch_counter_collection->callbacks.at(0)->buffer); EXPECT_EQ(*ctx.dispatch_counter_collection->callbacks.at(0)->buffer, opt_buff_id); + // Counter collection no longer registers a per-queue callback; activeness is observable via + // the context enabled flag (and counters::is_any_active()). bool found = false; ctx.dispatch_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.dispatch_counter_collection->callbacks.at(0)->queue_id) - { - found = true; - } - }); - EXPECT_TRUE(found); - /** * Check if context can be disabled correctly */ @@ -627,19 +620,12 @@ TEST(core, start_stop_callback_ctx) (void*) 0x54321); EXPECT_EQ(ctx.dispatch_counter_collection->callbacks.at(0)->context, get_client_ctx()); + // Counter collection no longer registers a per-queue callback; activeness is observable via + // the context enabled flag (and counters::is_any_active()). bool found = false; ctx.dispatch_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.dispatch_counter_collection->callbacks.at(0)->queue_id) - { - found = true; - } - }); - EXPECT_TRUE(found); - /** * Check if context can be disabled correctly */ diff --git a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/tests/per_agent_scoping_test.cpp b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/tests/per_agent_scoping_test.cpp new file mode 100644 index 000000000000..f05a6155424c --- /dev/null +++ b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/tests/per_agent_scoping_test.cpp @@ -0,0 +1,481 @@ +// MIT License +// +// Copyright (c) 2026 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/agent.hpp" +#include "lib/rocprofiler-sdk/context/context.hpp" +#include "lib/rocprofiler-sdk/counters/core.hpp" +#include "lib/rocprofiler-sdk/counters/queue_hooks.hpp" +#include "lib/rocprofiler-sdk/counters/tests/hsa_tables.hpp" +#include "lib/rocprofiler-sdk/hsa/agent_cache.hpp" +#include "lib/rocprofiler-sdk/hsa/queue.hpp" +#include "lib/rocprofiler-sdk/hsa/queue_controller.hpp" +#include "lib/rocprofiler-sdk/registration.hpp" + +#include +#include + +#include +#include + +#include + +using namespace rocprofiler::counters::test_constants; +using namespace rocprofiler; + +namespace rocprofiler +{ +namespace hsa +{ +class PerAgentFakeQueue : public Queue +{ +public: + PerAgentFakeQueue(const AgentCache& a, rocprofiler_queue_id_t id) + : Queue(a, get_api_table()) + , _agent(a) + , _id(id) + {} + const AgentCache& get_agent() const final { return _agent; }; + rocprofiler_queue_id_t get_id() const final { return _id; }; + + ~PerAgentFakeQueue() override = default; + +private: + const AgentCache& _agent; + rocprofiler_queue_id_t _id = {}; +}; +} // namespace hsa +} // namespace rocprofiler + +namespace +{ +// Two GPU agents are required to say anything about scoping: with one agent, "scoped to agent +// A" and "unscoped" are indistinguishable. +struct agent_pair +{ + const hsa::AgentCache* first = nullptr; + const hsa::AgentCache* second = nullptr; +}; + +void +test_init() +{ + static bool _initialized = false; + if(_initialized) return; + _initialized = true; + + HsaApiTable table; + table.amd_ext_ = &get_ext_table(); + table.core_ = &get_api_table(); + agent::construct_agent_cache(&table); + ASSERT_TRUE(hsa::get_queue_controller() != nullptr); + hsa::get_queue_controller()->init(get_api_table(), get_ext_table()); + registration::init_logging(); + registration::set_init_status(-1); + context::push_client(1); +} + +agent_pair +get_two_agents() +{ + auto result = agent_pair{}; + auto& agents = hsa::get_queue_controller()->get_supported_agents(); + for(const auto& [_, agent] : agents) + { + if(!agent.get_rocp_agent()) continue; + if(!result.first) + result.first = &agent; + else if(!result.second) + result.second = &agent; + else + break; + } + return result; +} + +void +noop_dispatch_cb(rocprofiler_dispatch_counting_service_data_t, + rocprofiler_counter_config_id_t*, + rocprofiler_user_data_t*, + void*) +{} + +void +noop_record_cb(rocprofiler_dispatch_counting_service_data_t, + rocprofiler_counter_record_t*, + size_t, + rocprofiler_user_data_t, + void*) +{} + +// Creates a context with a dispatch counting service configured on it. +rocprofiler_context_id_t +make_counter_context() +{ + auto ctx = rocprofiler_context_id_t{0}; + EXPECT_EQ(rocprofiler_create_context(&ctx), ROCPROFILER_STATUS_SUCCESS); + EXPECT_EQ(rocprofiler_configure_callback_dispatch_counting_service( + ctx, noop_dispatch_cb, nullptr, noop_record_cb, nullptr), + ROCPROFILER_STATUS_SUCCESS); + return ctx; +} + +// --------------------------------------------------------------------------------------------- +// rocprofiler_dispatch_counting_service_set_agents argument handling +// --------------------------------------------------------------------------------------------- + +TEST(counters_per_agent, set_agents_requires_a_configured_service) +{ + ASSERT_EQ(hsa_init(), HSA_STATUS_SUCCESS); + test_init(); + + auto ctx = rocprofiler_context_id_t{0}; + ASSERT_EQ(rocprofiler_create_context(&ctx), ROCPROFILER_STATUS_SUCCESS); + + auto agents = get_two_agents(); + if(!agents.first) GTEST_SKIP() << "no GPU agent available"; + auto agent_id = agents.first->get_rocp_agent()->id; + + // The agent set lives on the counter-collection service, so there is nothing to restrict + // until the service exists. Silently succeeding here would let a tool believe it had + // scoped collection when it had not. + EXPECT_EQ(rocprofiler_dispatch_counting_service_set_agents(ctx, &agent_id, 1), + ROCPROFILER_STATUS_ERROR_CONTEXT_NOT_FOUND); +} + +TEST(counters_per_agent, set_agents_rejects_bad_arguments) +{ + ASSERT_EQ(hsa_init(), HSA_STATUS_SUCCESS); + test_init(); + + auto agents = get_two_agents(); + if(!agents.first) GTEST_SKIP() << "no GPU agent available"; + + auto ctx = make_counter_context(); + auto agent_id = agents.first->get_rocp_agent()->id; + + EXPECT_EQ(rocprofiler_dispatch_counting_service_set_agents(ctx, nullptr, 1), + ROCPROFILER_STATUS_ERROR_INVALID_ARGUMENT); + + // A null array with no entries is the documented way to clear the restriction. + EXPECT_EQ(rocprofiler_dispatch_counting_service_set_agents(ctx, nullptr, 0), + ROCPROFILER_STATUS_SUCCESS); + + auto bogus = rocprofiler_agent_id_t{.handle = 0xDEADBEEF}; + EXPECT_EQ(rocprofiler_dispatch_counting_service_set_agents(ctx, &bogus, 1), + ROCPROFILER_STATUS_ERROR_AGENT_NOT_FOUND); + + // The rejected call must not have partially applied: a failed set leaves the previous + // (unrestricted) set in place rather than a half-built one. + auto* ctx_p = context::get_mutable_registered_context(ctx); + ASSERT_TRUE(ctx_p); + ASSERT_TRUE(ctx_p->dispatch_counter_collection); + EXPECT_TRUE(ctx_p->dispatch_counter_collection->agents.empty()); + + EXPECT_EQ(rocprofiler_dispatch_counting_service_set_agents(ctx, &agent_id, 1), + ROCPROFILER_STATUS_SUCCESS); + EXPECT_EQ(ctx_p->dispatch_counter_collection->agents.size(), 1U); + EXPECT_TRUE(ctx_p->dispatch_counter_collection->collects_on(agent_id)); +} + +TEST(counters_per_agent, set_agents_is_locked_while_the_context_is_active) +{ + ASSERT_EQ(hsa_init(), HSA_STATUS_SUCCESS); + test_init(); + + auto agents = get_two_agents(); + if(!agents.first) GTEST_SKIP() << "no GPU agent available"; + + auto ctx = make_counter_context(); + auto agent_id = agents.first->get_rocp_agent()->id; + + ASSERT_EQ(rocprofiler_start_context(ctx), ROCPROFILER_STATUS_SUCCESS); + + // The dispatch path reads the agent set without a lock, and start_context has already used + // it to scope serialization, so changing it mid-flight would silently desynchronize the two. + EXPECT_EQ(rocprofiler_dispatch_counting_service_set_agents(ctx, &agent_id, 1), + ROCPROFILER_STATUS_ERROR_CONFIGURATION_LOCKED); + + ASSERT_EQ(rocprofiler_stop_context(ctx), ROCPROFILER_STATUS_SUCCESS); + EXPECT_EQ(rocprofiler_dispatch_counting_service_set_agents(ctx, &agent_id, 1), + ROCPROFILER_STATUS_SUCCESS); +} + +// --------------------------------------------------------------------------------------------- +// Context conflict: the scenario from the #9586 review +// --------------------------------------------------------------------------------------------- + +// Before per-agent scoping this was impossible -- any second counter-collection context was +// rejected outright, so "context A on GPU-0, context B on GPU-1" could not even be expressed. +TEST(counters_per_agent, disjoint_agent_contexts_can_be_active_together) +{ + ASSERT_EQ(hsa_init(), HSA_STATUS_SUCCESS); + test_init(); + + auto agents = get_two_agents(); + if(!agents.second) GTEST_SKIP() << "fewer than two GPU agents available"; + + auto agent_a = agents.first->get_rocp_agent()->id; + auto agent_b = agents.second->get_rocp_agent()->id; + + auto ctx_a = make_counter_context(); + auto ctx_b = make_counter_context(); + + ASSERT_EQ(rocprofiler_dispatch_counting_service_set_agents(ctx_a, &agent_a, 1), + ROCPROFILER_STATUS_SUCCESS); + ASSERT_EQ(rocprofiler_dispatch_counting_service_set_agents(ctx_b, &agent_b, 1), + ROCPROFILER_STATUS_SUCCESS); + + EXPECT_EQ(rocprofiler_start_context(ctx_a), ROCPROFILER_STATUS_SUCCESS); + EXPECT_EQ(rocprofiler_start_context(ctx_b), ROCPROFILER_STATUS_SUCCESS); + + EXPECT_EQ(rocprofiler_stop_context(ctx_b), ROCPROFILER_STATUS_SUCCESS); + EXPECT_EQ(rocprofiler_stop_context(ctx_a), ROCPROFILER_STATUS_SUCCESS); +} + +TEST(counters_per_agent, overlapping_agent_contexts_still_conflict) +{ + ASSERT_EQ(hsa_init(), HSA_STATUS_SUCCESS); + test_init(); + + auto agents = get_two_agents(); + if(!agents.first) GTEST_SKIP() << "no GPU agent available"; + auto agent_a = agents.first->get_rocp_agent()->id; + + auto ctx_a = make_counter_context(); + auto ctx_b = make_counter_context(); + + ASSERT_EQ(rocprofiler_dispatch_counting_service_set_agents(ctx_a, &agent_a, 1), + ROCPROFILER_STATUS_SUCCESS); + ASSERT_EQ(rocprofiler_dispatch_counting_service_set_agents(ctx_b, &agent_a, 1), + ROCPROFILER_STATUS_SUCCESS); + + ASSERT_EQ(rocprofiler_start_context(ctx_a), ROCPROFILER_STATUS_SUCCESS); + EXPECT_EQ(rocprofiler_start_context(ctx_b), ROCPROFILER_STATUS_ERROR_CONTEXT_CONFLICT); + + ASSERT_EQ(rocprofiler_stop_context(ctx_a), ROCPROFILER_STATUS_SUCCESS); +} + +// An unrestricted context claims every agent, so it must keep conflicting with everything -- +// otherwise existing single-context tools would start silently sharing agents. +TEST(counters_per_agent, unrestricted_context_conflicts_with_a_scoped_one) +{ + ASSERT_EQ(hsa_init(), HSA_STATUS_SUCCESS); + test_init(); + + auto agents = get_two_agents(); + if(!agents.second) GTEST_SKIP() << "fewer than two GPU agents available"; + auto agent_b = agents.second->get_rocp_agent()->id; + + auto ctx_all = make_counter_context(); + auto ctx_scoped = make_counter_context(); + + ASSERT_EQ(rocprofiler_dispatch_counting_service_set_agents(ctx_scoped, &agent_b, 1), + ROCPROFILER_STATUS_SUCCESS); + + ASSERT_EQ(rocprofiler_start_context(ctx_all), ROCPROFILER_STATUS_SUCCESS); + EXPECT_EQ(rocprofiler_start_context(ctx_scoped), ROCPROFILER_STATUS_ERROR_CONTEXT_CONFLICT); + + ASSERT_EQ(rocprofiler_stop_context(ctx_all), ROCPROFILER_STATUS_SUCCESS); +} + +// --------------------------------------------------------------------------------------------- +// Serialization scope and reference counting +// --------------------------------------------------------------------------------------------- + +// The actual complaint in the review: a context that only wants GPU-1 must not serialize GPU-0. +TEST(counters_per_agent, serialization_is_scoped_to_the_contexts_agents) +{ + ASSERT_EQ(hsa_init(), HSA_STATUS_SUCCESS); + test_init(); + + auto agents = get_two_agents(); + if(!agents.second) GTEST_SKIP() << "fewer than two GPU agents available"; + + auto agent_a = agents.first->get_rocp_agent()->id; + auto agent_b = agents.second->get_rocp_agent()->id; + auto* controller = hsa::get_queue_controller(); + + // Touch both serializers so the agents are known to the controller; otherwise "not + // serialized" would be trivially true because the entry does not exist yet. + auto queue_a = hsa::PerAgentFakeQueue{*agents.first, {.handle = 101}}; + auto queue_b = hsa::PerAgentFakeQueue{*agents.second, {.handle = 102}}; + controller->serializer(&queue_a); + controller->serializer(&queue_b); + + auto ctx = make_counter_context(); + ASSERT_EQ(rocprofiler_dispatch_counting_service_set_agents(ctx, &agent_b, 1), + ROCPROFILER_STATUS_SUCCESS); + + auto* ctx_p = context::get_mutable_registered_context(ctx); + ASSERT_TRUE(ctx_p && ctx_p->dispatch_counter_collection); + ASSERT_EQ(ctx_p->dispatch_counter_collection->agents.size(), 1U) << "agent set not applied"; + ASSERT_TRUE(ctx_p->dispatch_counter_collection->collects_on(agent_b)); + ASSERT_FALSE(ctx_p->dispatch_counter_collection->collects_on(agent_a)); + + EXPECT_FALSE(controller->is_serialization_enabled(agent_a)); + EXPECT_FALSE(controller->is_serialization_enabled(agent_b)); + + ASSERT_EQ(rocprofiler_start_context(ctx), ROCPROFILER_STATUS_SUCCESS); + + EXPECT_TRUE(controller->is_serialization_enabled(agent_b)) + << "the context's own agent must be serialized"; + EXPECT_FALSE(controller->is_serialization_enabled(agent_a)) + << "an agent outside the context's agent set must not be serialized"; + + ASSERT_EQ(rocprofiler_stop_context(ctx), ROCPROFILER_STATUS_SUCCESS); + EXPECT_FALSE(controller->is_serialization_enabled(agent_b)); +} + +// Counter collection, thread trace and SPM each enable serialization independently. Without +// reference counting the first one to stop unserializes the others, which is a pre-existing +// bug that per-agent scoping would otherwise make easier to hit. +TEST(counters_per_agent, serialization_is_reference_counted_across_subsystems) +{ + ASSERT_EQ(hsa_init(), HSA_STATUS_SUCCESS); + test_init(); + + auto agents = get_two_agents(); + if(!agents.first) GTEST_SKIP() << "no GPU agent available"; + + auto agent_a = agents.first->get_rocp_agent()->id; + auto* controller = hsa::get_queue_controller(); + + auto queue_a = hsa::PerAgentFakeQueue{*agents.first, {.handle = 201}}; + controller->serializer(&queue_a); + + ASSERT_FALSE(controller->is_serialization_enabled(agent_a)); + + controller->enable_serialization(); + controller->enable_serialization(); + EXPECT_TRUE(controller->is_serialization_enabled(agent_a)); + + controller->disable_serialization(); + EXPECT_TRUE(controller->is_serialization_enabled(agent_a)) + << "one subsystem stopping must not unserialize an agent another still needs"; + + controller->disable_serialization(); + EXPECT_FALSE(controller->is_serialization_enabled(agent_a)); + + // Over-releasing must not drive the count negative, or the next enable would be swallowed. + controller->disable_serialization(); + controller->enable_serialization(); + EXPECT_TRUE(controller->is_serialization_enabled(agent_a)); + controller->disable_serialization(); + EXPECT_FALSE(controller->is_serialization_enabled(agent_a)); +} + +// A scoped request and an unscoped request overlap on the scoped agent; releasing one must not +// release the other. +TEST(counters_per_agent, scoped_and_unscoped_serialization_compose) +{ + ASSERT_EQ(hsa_init(), HSA_STATUS_SUCCESS); + test_init(); + + auto agents = get_two_agents(); + if(!agents.second) GTEST_SKIP() << "fewer than two GPU agents available"; + + auto agent_a = agents.first->get_rocp_agent()->id; + auto agent_b = agents.second->get_rocp_agent()->id; + auto* controller = hsa::get_queue_controller(); + + auto queue_a = hsa::PerAgentFakeQueue{*agents.first, {.handle = 301}}; + auto queue_b = hsa::PerAgentFakeQueue{*agents.second, {.handle = 302}}; + controller->serializer(&queue_a); + controller->serializer(&queue_b); + + controller->enable_serialization({agent_b}); + EXPECT_FALSE(controller->is_serialization_enabled(agent_a)); + EXPECT_TRUE(controller->is_serialization_enabled(agent_b)); + + controller->enable_serialization(); + EXPECT_TRUE(controller->is_serialization_enabled(agent_a)); + EXPECT_TRUE(controller->is_serialization_enabled(agent_b)); + + controller->disable_serialization(); + EXPECT_FALSE(controller->is_serialization_enabled(agent_a)); + EXPECT_TRUE(controller->is_serialization_enabled(agent_b)) + << "the scoped request still holds agent B"; + + controller->disable_serialization({agent_b}); + EXPECT_FALSE(controller->is_serialization_enabled(agent_b)); +} + +// --------------------------------------------------------------------------------------------- +// Dispatch path +// --------------------------------------------------------------------------------------------- + +// queue_cb returns serialize=true even when it declines to instrument, so a tool cannot avoid +// serializing a GPU by returning a null config -- the filtering has to happen before the call. +// This is the end-to-end form of the review's example. +TEST(counters_per_agent, enter_hook_ignores_dispatches_on_other_agents) +{ + ASSERT_EQ(hsa_init(), HSA_STATUS_SUCCESS); + test_init(); + + auto agents = get_two_agents(); + if(!agents.second) GTEST_SKIP() << "fewer than two GPU agents available"; + + auto agent_b = agents.second->get_rocp_agent()->id; + + auto ctx = make_counter_context(); + ASSERT_EQ(rocprofiler_dispatch_counting_service_set_agents(ctx, &agent_b, 1), + ROCPROFILER_STATUS_SUCCESS); + ASSERT_EQ(rocprofiler_start_context(ctx), ROCPROFILER_STATUS_SUCCESS); + + auto packet = hsa::rocprofiler_packet{}; + auto corr_id = context::correlation_id{}; + corr_id.internal = 4242; + + // A dispatch on the agent the context does not collect on. + { + auto queue_a = hsa::PerAgentFakeQueue{*agents.first, {.handle = 401}}; + auto inst_pkt = hsa::inst_pkt_t{}; + bool is_serialized = false; + auto user_data = rocprofiler_user_data_t{.value = corr_id.internal}; + + counters::kernel_dispatch_phase_enter_hook( + queue_a, packet, 42, 1, &user_data, {}, &corr_id, inst_pkt, is_serialized); + + EXPECT_FALSE(is_serialized) + << "a dispatch on an agent outside the context's agent set must not be serialized"; + EXPECT_TRUE(inst_pkt.empty()) + << "a dispatch on an agent outside the context's agent set must not be instrumented"; + } + + // The same dispatch on the agent the context does collect on still goes through, so the + // test cannot pass just because the hook stopped doing anything at all. + { + auto queue_b = hsa::PerAgentFakeQueue{*agents.second, {.handle = 402}}; + auto inst_pkt = hsa::inst_pkt_t{}; + bool is_serialized = false; + auto user_data = rocprofiler_user_data_t{.value = corr_id.internal}; + + counters::kernel_dispatch_phase_enter_hook( + queue_b, packet, 42, 2, &user_data, {}, &corr_id, inst_pkt, is_serialized); + + EXPECT_TRUE(is_serialized) + << "a dispatch on the context's own agent must still be serialized"; + } + + ASSERT_EQ(rocprofiler_stop_context(ctx), ROCPROFILER_STATUS_SUCCESS); +} +} // namespace diff --git a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/tests/queue_hooks_test.cpp b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/tests/queue_hooks_test.cpp new file mode 100644 index 000000000000..716cb2cdf0a6 --- /dev/null +++ b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/counters/tests/queue_hooks_test.cpp @@ -0,0 +1,388 @@ +// MIT License +// +// Copyright (c) 2026 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/counters/queue_hooks.hpp" +#include "lib/rocprofiler-sdk/agent.hpp" +#include "lib/rocprofiler-sdk/context/context.hpp" +#include "lib/rocprofiler-sdk/counters/core.hpp" +#include "lib/rocprofiler-sdk/counters/dispatch_handlers.hpp" +#include "lib/rocprofiler-sdk/counters/tests/hsa_tables.hpp" +#include "lib/rocprofiler-sdk/hsa/agent_cache.hpp" +#include "lib/rocprofiler-sdk/hsa/aql_packet.hpp" +#include "lib/rocprofiler-sdk/hsa/queue.hpp" +#include "lib/rocprofiler-sdk/hsa/queue_controller.hpp" +#include "lib/rocprofiler-sdk/hsa/queue_hooks/client_ids.hpp" +#include "lib/rocprofiler-sdk/kernel_dispatch/profiling_time.hpp" +#include "lib/rocprofiler-sdk/registration.hpp" + +#include +#include + +#include +#include + +#include +#include +#include +#include + +using namespace rocprofiler::counters::test_constants; +using namespace rocprofiler; + +#define ROCPROFILER_CALL(result, msg) \ + { \ + rocprofiler_status_t CHECKSTATUS = result; \ + ASSERT_EQ(CHECKSTATUS, ROCPROFILER_STATUS_SUCCESS) \ + << msg << ": " << rocprofiler_get_status_string(CHECKSTATUS); \ + } + +namespace rocprofiler +{ +namespace hsa +{ +class QueueHooksFakeQueue : public Queue +{ +public: + QueueHooksFakeQueue(const AgentCache& a, rocprofiler_queue_id_t id) + : Queue(a, get_api_table()) + , _agent(a) + , _id(id) + {} + const AgentCache& get_agent() const final { return _agent; }; + rocprofiler_queue_id_t get_id() const final { return _id; }; + + ~QueueHooksFakeQueue() override = default; + +private: + const AgentCache& _agent; + rocprofiler_queue_id_t _id = {}; +}; +} // namespace hsa +} // namespace rocprofiler + +namespace +{ +rocprofiler_context_id_t& +get_client_ctx() +{ + static rocprofiler_context_id_t ctx{0}; + return ctx; +} + +void +test_init() +{ + HsaApiTable table; + table.amd_ext_ = &get_ext_table(); + table.core_ = &get_api_table(); + agent::construct_agent_cache(&table); + ASSERT_TRUE(hsa::get_queue_controller() != nullptr); + hsa::get_queue_controller()->init(get_api_table(), get_ext_table()); +} + +struct expected_dispatch +{ + rocprofiler_counter_config_id_t id = {.handle = 0}; + rocprofiler_queue_id_t queue_id = {.handle = 0}; + rocprofiler_agent_id_t agent_id = {.handle = 0}; + uint64_t kernel_id = 0; + uint64_t dispatch_id = 0; +}; + +void +user_dispatch_cb(rocprofiler_dispatch_counting_service_data_t dispatch_data, + rocprofiler_counter_config_id_t* config, + rocprofiler_user_data_t*, + void* callback_data_args) +{ + auto& expected = *static_cast(callback_data_args); + ASSERT_NE(config, nullptr); + config->handle = expected.id.handle; + (void) dispatch_data; +} + +// Records delivered to the tool, keyed by dispatch id. Guarded because before stop_context the +// consumer thread delivers these, and after it consumer_thread_t::add() consumes inline on the +// caller -- so the writer can be either thread depending on when the completion lands. +struct delivered_records +{ + std::mutex mtx{}; + std::vector dispatch_ids{}; + + void add(size_t dispatch_id) + { + auto lk = std::unique_lock{mtx}; + dispatch_ids.emplace_back(dispatch_id); + } + + std::vector snapshot() + { + auto lk = std::unique_lock{mtx}; + return dispatch_ids; + } + + // Reset so a repeated run of the test does not see the previous run's deliveries. + void clear() + { + auto lk = std::unique_lock{mtx}; + dispatch_ids.clear(); + } +}; + +delivered_records& +get_delivered() +{ + static delivered_records _v{}; + return _v; +} + +void +user_record_cb(rocprofiler_dispatch_counting_service_data_t dispatch_data, + rocprofiler_counter_record_t*, + size_t, + rocprofiler_user_data_t, + void*) +{ + get_delivered().add(dispatch_data.dispatch_info.dispatch_id); +} + +// is_any_active replaces the queue.get_notifiers() signal that the old per-queue +// callback registration used to provide to the HSA write interceptor gate. With +// no dispatch-counter-collection context active, it must report inactive. This +// requires no GPU / HSA runtime. +TEST(counters_queue_hooks, is_any_active_false_when_no_context_active) +{ + EXPECT_FALSE(rocprofiler::counters::is_any_active()); +} + +TEST(counters_queue_hooks, exit_hook_skips_when_inst_pkt_has_no_counter_client_id) +{ + ASSERT_EQ(hsa_init(), HSA_STATUS_SUCCESS); + test_init(); + + auto agents = hsa::get_queue_controller()->get_supported_agents(); + ASSERT_FALSE(agents.empty()); + const auto& [_, agent] = *agents.begin(); + hsa::QueueHooksFakeQueue fq(agent, {.handle = 1}); + + hsa::inst_pkt_t inst_pkt; + inst_pkt.emplace_back(std::make_pair(std::make_unique(), + rocprofiler::hsa::queue_hooks::THREAD_TRACE_CLIENT_ID)); + + // The hook returns before it dereferences the session, so a null one keeps this test free of + // any HSA runtime: queue_info_session_t holds a Queue& and cannot be default constructed. + auto sess = std::shared_ptr{}; + auto packet = rocprofiler::hsa::packet_data_t{}; + auto fq_pkt = rocprofiler::hsa::rocprofiler_packet{}; + + // Must return without touching registered counter contexts. + rocprofiler::counters::kernel_dispatch_phase_exit_hook( + nullptr, fq_pkt, sess, packet, inst_pkt, rocprofiler::kernel_dispatch::profiling_time{}); + SUCCEED(); +} + +TEST(counters_queue_hooks, is_any_active_true_while_context_started) +{ + ASSERT_EQ(hsa_init(), HSA_STATUS_SUCCESS); + test_init(); + + registration::init_logging(); + registration::set_init_status(-1); + context::push_client(1); + + ROCPROFILER_CALL(rocprofiler_create_context(&get_client_ctx()), "context creation failed"); + ROCPROFILER_CALL(rocprofiler_configure_callback_dispatch_counting_service( + get_client_ctx(), user_dispatch_cb, nullptr, nullptr, nullptr), + "Could not setup counting service"); + ROCPROFILER_CALL(rocprofiler_start_context(get_client_ctx()), "start context"); + + EXPECT_TRUE(rocprofiler::counters::is_any_active()); + + ROCPROFILER_CALL(rocprofiler_stop_context(get_client_ctx()), "stop context"); + EXPECT_FALSE(rocprofiler::counters::is_any_active()); + + registration::set_init_status(1); + registration::finalize(); + context::pop_client(1); + get_client_ctx() = rocprofiler_context_id_t{0}; +} + +// Regression for callback-registry removal, per the review on #8891: dispatches enqueued while the +// context is active must still complete when stop_context runs before their GPU completions land. +// +// Two invariants are checked, because they fail independently: +// +// 1. Routing. Each dispatch's packet_return_map entry is erased by its own completion, so the map +// is asserted to drop by exactly one per completion rather than only to reach zero at the end. +// An aggregate check would pass even if one completion erased another dispatch's entry. +// 2. Delivery. Every dispatch reaches the tool's record callback. Erasure happens in completed_cb +// before the hand-off to the consumer, so a test that only checks the map would pass even if +// nothing were ever delivered -- which is the failure the review actually described. +// +// Several dispatches are in flight at once, since a single one cannot distinguish "routed the one +// entry" from "routed by provenance". +// +// No kernel executes here, so the counter values are whatever the profile's output buffer holds; +// the assertion is about delivery, not values. +TEST(counters_queue_hooks, stop_context_in_flight_completion_routes_via_hook_path) +{ + constexpr size_t num_dispatches = 3; + + ASSERT_EQ(hsa_init(), HSA_STATUS_SUCCESS); + test_init(); + + registration::init_logging(); + registration::set_init_status(-1); + context::push_client(1); + + ROCPROFILER_CALL(rocprofiler_create_context(&get_client_ctx()), "context creation failed"); + // A record callback is supplied so delivery is observable; the base test passed nullptr, which + // is why it could not detect a dropped completion. + ROCPROFILER_CALL(rocprofiler_configure_callback_dispatch_counting_service( + get_client_ctx(), user_dispatch_cb, nullptr, user_record_cb, nullptr), + "Could not setup counting service"); + ROCPROFILER_CALL(rocprofiler_start_context(get_client_ctx()), "start context"); + + auto* ctx_p = context::get_mutable_registered_context(get_client_ctx()); + ASSERT_TRUE(ctx_p); + ASSERT_TRUE(ctx_p->dispatch_counter_collection); + ASSERT_EQ(ctx_p->dispatch_counter_collection->callbacks.size(), 1); + auto cb_info = ctx_p->dispatch_counter_collection->callbacks.front(); + + ASSERT_TRUE(hsa::get_queue_controller() != nullptr); + auto agents = hsa::get_queue_controller()->get_supported_agents(); + ASSERT_GT(agents.size(), 0); + + const auto& [_, agent] = *agents.begin(); + ASSERT_TRUE(agent.get_rocp_agent()); + + rocprofiler_counter_id_t counter_id = {.handle = 0}; + { + auto mets = rocprofiler::counters::loadMetrics(); + const auto* gfx_metrics = + rocprofiler::common::get_val(mets->arch_to_metric, std::string(agent.name())); + ASSERT_TRUE(gfx_metrics); + ASSERT_FALSE(gfx_metrics->empty()); + for(const auto& metric : *gfx_metrics) + { + if(metric.expression().empty()) + { + counter_id.handle = metric.id(); + break; + } + } + } + ASSERT_NE(counter_id.handle, 0U); + + expected_dispatch expected = {}; + ROCPROFILER_CALL( + rocprofiler_create_counter_config(agent.get_rocp_agent()->id, &counter_id, 1, &expected.id), + "Unable to create profile"); + cb_info->callback_args = &expected; + expected.queue_id = {.handle = 1}; + expected.agent_id = agent.get_rocp_agent()->id; + expected.kernel_id = 42; + expected.dispatch_id = 7; + + hsa::QueueHooksFakeQueue fq(agent, expected.queue_id); + hsa::rocprofiler_packet pkt{}; + // correlation_id declares constructors and holds private ref counters, so it is not an + // aggregate and cannot take a designated initializer. + context::correlation_id corr_id{}; + corr_id.internal = 99; + + get_delivered().clear(); + + // Enqueue several dispatches while the context is active. Each queue_cb call must register its + // own entry; if the implementation ever pooled one packet across concurrent dispatches, the + // size assertion below would catch it rather than the test silently degenerating to one. + std::vector> in_flight; + std::vector dispatch_ids; + for(size_t i = 0; i < num_dispatches; ++i) + { + const size_t dispatch_id = expected.dispatch_id + i; + auto user_data = rocprofiler_user_data_t{.value = corr_id.internal}; + auto ret_pkt = rocprofiler::counters::queue_cb( + ctx_p, cb_info, fq, pkt, expected.kernel_id, dispatch_id, &user_data, {}, &corr_id); + ASSERT_TRUE(ret_pkt.packet) << "queue_cb produced no instrumentation for dispatch " << i; + in_flight.emplace_back(std::move(ret_pkt.packet)); + dispatch_ids.emplace_back(dispatch_id); + } + + size_t map_size_before_stop = 0; + cb_info->packet_return_map.rlock([&](const auto& data) { map_size_before_stop = data.size(); }); + ASSERT_EQ(map_size_before_stop, num_dispatches) + << "queue_cb should register one packet_return_map entry per in-flight dispatch"; + + // Stop while every dispatch is still in flight on the GPU. This is the window the review + // described: the context leaves the active list, so a completion hook that routed by + // activeness would find nothing to deliver to. + ROCPROFILER_CALL(rocprofiler_stop_context(get_client_ctx()), "stop context"); + EXPECT_FALSE(rocprofiler::counters::is_any_active()); + + // Complete them one at a time, checking the map drops by exactly one each time. + for(size_t i = 0; i < num_dispatches; ++i) + { + auto sess = + std::make_shared(hsa::queue_info_session_t{.queue = fq}); + auto& packet_data = sess->packet_data.emplace_back(); + sess->correlation_id = &corr_id; + packet_data.callback_record.dispatch_info.dispatch_id = dispatch_ids[i]; + + rocprofiler::counters::inst_pkt_t inst_pkt; + // Construct the pair in place rather than via make_pair: make_pair would deduce the tag's + // enum type and rely on a pair-to-pair conversion, where piecewise construction converts + // the tag to ClientID directly. Also matches how the production hook emplaces. + inst_pkt.emplace_back(std::move(in_flight[i]), hsa::queue_hooks::COUNTERS_CLIENT_ID); + + rocprofiler::counters::kernel_dispatch_phase_exit_hook( + &fq, pkt, sess, packet_data, inst_pkt, rocprofiler::kernel_dispatch::profiling_time{}); + + size_t remaining = num_dispatches; + cb_info->packet_return_map.rlock([&](const auto& data) { remaining = data.size(); }); + EXPECT_EQ(remaining, num_dispatches - (i + 1)) + << "completion " << i + << " should erase exactly its own packet_return_map entry after stop_context"; + } + + size_t map_size_after_completion = num_dispatches; + cb_info->packet_return_map.rlock( + [&](const auto& data) { map_size_after_completion = data.size(); }); + EXPECT_EQ(map_size_after_completion, 0U) + << "packet_return_map must drain via kernel_dispatch_phase_exit_hook after stop_context"; + + // Delivery. callback_thread_stop() has already run as part of stop_context, so these arrive + // through the inline path in consumer_thread_t::add() rather than the consumer thread -- which + // is precisely the lifetime property the review asked to be kept intact. + auto delivered = get_delivered().snapshot(); + EXPECT_EQ(delivered.size(), num_dispatches) + << "every in-flight dispatch must reach the tool's record callback after stop_context"; + for(auto dispatch_id : dispatch_ids) + { + EXPECT_NE(std::find(delivered.begin(), delivered.end(), dispatch_id), delivered.end()) + << "no record delivered for dispatch " << dispatch_id; + } + + registration::set_init_status(1); + registration::finalize(); +} +} // namespace diff --git a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/dispatch_counting_service.cpp b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/dispatch_counting_service.cpp index 6631fa7b3c92..4f74ef5efa58 100644 --- a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/dispatch_counting_service.cpp +++ b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/dispatch_counting_service.cpp @@ -75,4 +75,20 @@ rocprofiler_configure_callback_dispatch_counting_service( record_callback, record_callback_args); } + +/** + * @brief Restrict a dispatch counting service to a set of GPU agents. + * + * @param [in] context_id context id with a configured dispatch counting service + * @param [in] agents array of GPU agent ids to restrict collection to + * @param [in] num_agents number of entries in agents + * @return ::rocprofiler_status_t + */ +rocprofiler_status_t +rocprofiler_dispatch_counting_service_set_agents(rocprofiler_context_id_t context_id, + const rocprofiler_agent_id_t* agents, + size_t num_agents) +{ + return rocprofiler::counters::set_dispatch_agents(context_id, agents, num_agents); +} } diff --git a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/hsa/CMakeLists.txt b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/hsa/CMakeLists.txt index ab27514b1822..c085f2c7f7cd 100644 --- a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/hsa/CMakeLists.txt +++ b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/hsa/CMakeLists.txt @@ -39,3 +39,4 @@ target_sources(rocprofiler-sdk-object-library PRIVATE ${ROCPROFILER_LIB_HSA_SOUR ${ROCPROFILER_LIB_HSA_HEADERS}) add_subdirectory(details) +add_subdirectory(queue_hooks) diff --git a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/hsa/queue.cpp b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/hsa/queue.cpp index 36e67f1e642e..6cce070b81b8 100644 --- a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/hsa/queue.cpp +++ b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/hsa/queue.cpp @@ -29,6 +29,7 @@ #include "lib/common/utility.hpp" #include "lib/rocprofiler-sdk/code_object/code_object.hpp" #include "lib/rocprofiler-sdk/context/context.hpp" +#include "lib/rocprofiler-sdk/counters/queue_hooks.hpp" #include "lib/rocprofiler-sdk/hip/event.hpp" #include "lib/rocprofiler-sdk/hip/graph.hpp" #include "lib/rocprofiler-sdk/hsa/details/fmt.hpp" @@ -294,6 +295,15 @@ AsyncSignalHandler(hsa_signal_value_t /*signal_v*/, void* data) } }); + // Counter collection completion is migrated off the callback registry (see + // WriteInterceptor); invoke it explicitly here. + counters::kernel_dispatch_phase_exit_hook(&queue_info_session.queue, + packet.kernel_packet, + _session, + packet, + packet.instrumentation_packets, + dispatch_time); + CHECK_NOTNULL(hsa::get_queue_controller()) ->serializer(&queue_info_session.queue) .wlock([&](auto& serializer) { @@ -428,8 +438,12 @@ WriteInterceptor(const void* packets, auto* gls = ::rocprofiler::hip::graph::current_launch_state(); const bool graph_launch_active = (gls != nullptr); + const bool counters_active = counters::is_any_active(); + // Counter collection no longer registers a queue-controller callback, so it does not count + // toward get_notifiers(); detect it explicitly so a counters-only run still enters the + // interceptor. const bool no_real_consumers = - (queue.get_notifiers() == 0 && + (queue.get_notifiers() == 0 && !counters_active && context::get_active_contexts(full_packet_instrumentation_context_filter).empty()); const bool has_kernel_replay = kernel_replay::has_active_replay_contexts(); @@ -827,6 +841,19 @@ WriteInterceptor(const void* packets, } }); + // Counter collection is migrated off the per-queue callback registry: call its hook + // explicitly (the other services still flow through signal_callback above). + counters::kernel_dispatch_phase_enter_hook( + queue, + kernel_packet, + kernel_id, + dispatch_id, + &_packet_data.user_data, + _packet_data.tracing_data.external_correlation_ids, + corr_id, + _packet_data.instrumentation_packets, + _packet_data.is_serialized); + bool inserted_before = false; if(_packet_data.is_serialized) { @@ -1195,6 +1222,9 @@ WriteInterceptor(const void* packets, } }); + // Counter collection requires per-packet mode; it no longer participates in the registry. + if(counters_active) should_batch_packets = false; + if(should_batch_packets) { ROCP_TRACE_IF(pkt_count > 1) << fmt::format( diff --git a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/hsa/queue_controller.cpp b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/hsa/queue_controller.cpp index 4b334e34fd1b..27625b0532ce 100644 --- a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/hsa/queue_controller.cpp +++ b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/hsa/queue_controller.cpp @@ -677,26 +677,36 @@ common::Synchronized& QueueController::serializer(const Queue* queue) { CHECK(queue); + const auto agent_id = queue->get_agent().get_rocp_agent()->id; + common::Synchronized* ret = nullptr; - _profiler_serializer.ulock( - [&](const auto& m) { - if(auto ptr = m.find(queue->get_agent().get_rocp_agent()->id); ptr != m.end()) - { - ret = ptr->second.get(); + // Hold the refcount read lock across the lookup/insertion so that a concurrent + // update_serialization() cannot run between reading the refcount and inserting the + // entry: if it did, it would find no serializer to transition and the new entry would + // be created in the wrong state. update_serialization() holds the refcount write lock + // then the serializer write lock, so taking them in the same order here is deadlock-free. + _serialization_refcount.rlock([&](const auto& state) { + const bool should_serialize = state.enabled(agent_id); + _profiler_serializer.ulock( + [&](const auto& m) { + if(auto ptr = m.find(agent_id); ptr != m.end()) + { + ret = ptr->second.get(); + return true; + } + return false; + }, + [&](auto& m) { + ret = m.emplace(agent_id, + std::make_shared>()) + .first->second.get(); + if(should_serialize) + { + ret->wlock([&](auto& serializer) { serializer.enable({}); }); + } return true; - } - return false; - }, - [&](auto& m) { - ret = m.emplace(queue->get_agent().get_rocp_agent()->id, - std::make_shared>()) - .first->second.get(); - if(_serialized_enabled.load() == true) - { - ret->wlock([&](auto& serializer) { serializer.enable({}); }); - } - return true; - }); + }); + }); return *ret; } @@ -715,47 +725,107 @@ per_dev_map(const QueueController::queue_map_t& _queues_v) }; // namespace void -QueueController::disable_serialization() +QueueController::update_serialization(const agent_handle_set_t& agents, bool enable) { _queues.rlock([&](const queue_map_t& _queues_v) { - _serialized_enabled.store(false); auto pd_map = per_dev_map(_queues_v); - _profiler_serializer.wlock([&](auto& m) { - for(auto& [k, v] : m) + + // Agents whose effective refcount crossed zero in this call; only those get their + // serializer toggled. Both refcount mutation and serializer transitions happen + // together inside the refcount lock to prevent concurrent calls from interleaving. + bool any_enabled = false; + + _serialization_refcount.wlock([&](auto& state) { + // An agent can be covered by `all` and by an explicit entry at the same time, so + // the transition has to be decided by comparing effective counts across the + // update rather than by watching a single counter hit zero. + auto was_enabled = std::unordered_map{}; + _profiler_serializer.rlock([&](const auto& serializers) { + for(const auto& [agent_id, _] : serializers) + was_enabled.emplace(agent_id, state.enabled(agent_id)); + }); + + if(agents.empty()) + { + if(enable) + ++state.all; + else if(state.all > 0) + --state.all; + } + else { - if(auto it = pd_map.find(k); it != pd_map.end()) + for(const auto& agent_id : agents) { - v->wlock([&](auto& serializer) { serializer.disable(it->second); }); + auto& count = state.per_agent[agent_id]; + if(enable) + ++count; + else if(count > 0) + --count; } - else + } + + _profiler_serializer.wlock([&](auto& serializers) { + for(const auto& [agent_id, _] : serializers) { - v->wlock([&](auto& serializer) { serializer.disable({}); }); + auto itr = was_enabled.find(agent_id); + if(itr == was_enabled.end()) continue; + const bool now_enabled = state.enabled(agent_id); + if(itr->second == now_enabled) continue; + + auto queues = hsa_barrier::queue_map_ptr_t{}; + if(auto it = pd_map.find(agent_id); it != pd_map.end()) queues = it->second; + + auto ser_itr = serializers.find(agent_id); + if(ser_itr == serializers.end() || !ser_itr->second) continue; + // Derived from the post-update refcount rather than the caller's `enable` + // flag, so the applied state cannot disagree with the count that produced + // the transition. + ser_itr->second->wlock([&](auto& serializer) { + if(now_enabled) + serializer.enable(queues); + else + serializer.disable(queues); + }); } - } + }); + + any_enabled = state.any(); }); + + _serialized_enabled.store(any_enabled); }); } +void +QueueController::disable_serialization() +{ + update_serialization({}, false); +} + void QueueController::enable_serialization() { - _queues.rlock([&](const queue_map_t& _queues_v) { - _serialized_enabled.store(true); - auto pd_map = per_dev_map(_queues_v); - _profiler_serializer.wlock([&](auto& m) { - for(auto& [k, v] : m) - { - if(auto it = pd_map.find(k); it != pd_map.end()) - { - v->wlock([&](auto& serializer) { serializer.enable(it->second); }); - } - else - { - v->wlock([&](auto& serializer) { serializer.enable({}); }); - } - } - }); - }); + update_serialization({}, true); +} + +void +QueueController::disable_serialization(const agent_handle_set_t& agents) +{ + update_serialization(agents, false); +} + +void +QueueController::enable_serialization(const agent_handle_set_t& agents) +{ + update_serialization(agents, true); +} + +bool +QueueController::is_serialization_enabled(rocprofiler_agent_id_t agent_id) const +{ + bool enabled = false; + _serialization_refcount.rlock([&](const auto& state) { enabled = state.enabled(agent_id); }); + return enabled; } void diff --git a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/hsa/queue_controller.hpp b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/hsa/queue_controller.hpp index b42ae02394f8..c1cbf4936480 100644 --- a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/hsa/queue_controller.hpp +++ b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/hsa/queue_controller.hpp @@ -30,11 +30,13 @@ #include #include +#include #include #include #include #include +#include #include namespace rocprofiler @@ -100,13 +102,27 @@ class QueueController common::Synchronized& serializer(const Queue*); + // An empty set means every GPU agent. + using agent_handle_set_t = std::unordered_set; + /** - * 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. + * Enable/disable serialization for QueueController, has no effect if counter collection + * is not in use (which defaults to no serialization mechanism). + * + * Serialization is reference counted per agent. Counter collection, thread trace and SPM + * each enable it independently, so an agent stays serialized until every subsystem that + * asked for it has released it -- otherwise whichever service stops first would silently + * unserialize the ones still running. The no-argument overloads apply to every GPU agent, + * which is the behavior every caller had before agents could be scoped. */ void enable_serialization(); void disable_serialization(); + void enable_serialization(const agent_handle_set_t& agents); + void disable_serialization(const agent_handle_set_t& agents); + + // Whether the given agent currently has serialization enabled. Exposed for tests and for + // callers that need to reason about scope rather than trigger a transition. + bool is_serialization_enabled(rocprofiler_agent_id_t agent_id) const; // Prints current state of signals for queues, used for debugging. Only prints // serialization related signals if not compiled in debug mode. @@ -132,6 +148,39 @@ class QueueController std::unordered_map>>> _profiler_serializer; + // How many subsystems currently want serialization, per agent. `all` counts the callers + // that asked for every agent; it is kept separate from `per_agent` because an agent's + // serializer is created lazily on first use and an unscoped request has to cover agents + // that do not exist yet. + struct serialization_refcount + { + int64_t all = 0; + std::unordered_map per_agent = {}; + + int64_t count(rocprofiler_agent_id_t agent_id) const + { + auto itr = per_agent.find(agent_id); + return all + ((itr == per_agent.end()) ? 0 : itr->second); + } + + bool enabled(rocprofiler_agent_id_t agent_id) const { return count(agent_id) > 0; } + + bool any() const + { + if(all > 0) return true; + for(const auto& [agent_id, count] : per_agent) + { + if(count > 0) return true; + } + return false; + } + }; + + common::Synchronized _serialization_refcount; + + // Applies a +1/-1 to the refcount of each agent in `agents` (empty == all GPU agents) and + // toggles only the serializers whose effective count crossed zero. + void update_serialization(const agent_handle_set_t& agents, bool enable); }; QueueController* diff --git a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/hsa/queue_hooks/CMakeLists.txt b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/hsa/queue_hooks/CMakeLists.txt new file mode 100644 index 000000000000..5f5e8f8481f2 --- /dev/null +++ b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/hsa/queue_hooks/CMakeLists.txt @@ -0,0 +1,9 @@ +# +# Shared, header-only client-id tags used by the per-service queue hooks. +# +set(ROCPROFILER_LIB_HSA_QUEUE_HOOKS_SOURCES) +set(ROCPROFILER_LIB_HSA_QUEUE_HOOKS_HEADERS client_ids.hpp) + +target_sources( + rocprofiler-sdk-object-library PRIVATE ${ROCPROFILER_LIB_HSA_QUEUE_HOOKS_SOURCES} + ${ROCPROFILER_LIB_HSA_QUEUE_HOOKS_HEADERS}) diff --git a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/hsa/queue_hooks/client_ids.hpp b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/hsa/queue_hooks/client_ids.hpp new file mode 100644 index 000000000000..b7a45802fef9 --- /dev/null +++ b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk/hsa/queue_hooks/client_ids.hpp @@ -0,0 +1,64 @@ +// MIT License +// +// Copyright (c) 2026 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 + +namespace rocprofiler +{ +namespace hsa +{ +namespace queue_hooks +{ +// Tags identifying the producer subsystem of each inst_pkt_t entry, stored in the +// inst_pkt_t pair as an hsa::ClientID. +// +// The underlying type is fixed to int64_t to match hsa::ClientID. The enum is +// deliberately unscoped rather than an enum class: unscoped keeps the implicit +// conversion to ClientID, so the emplace sites need no cast, while still grouping +// the values under one type. +// +// The values are negative because they share the inst_pkt_t tag space with the +// per-queue callback registry, whose QueueController::add_callback hands out +// strictly positive ids counting up from 1. Migrated services (counters, thread +// trace, SPM) select their own packets by comparing against these tags, so a tag +// drawn from the registry's range could match a packet the service does not own. +// Services that have not migrated yet still receive registry ids, so the two +// ranges must stay disjoint until the registry is gone. Any tag added here must +// likewise be negative; -1 is left unused because QueueController::add_callback +// uses it as a local "unassigned" initializer. +enum client_id : int64_t +{ + COUNTERS_CLIENT_ID = -1001, + THREAD_TRACE_CLIENT_ID = -1002, + PC_SAMPLING_CLIENT_ID = -1003, + SPM_CLIENT_ID = -1004, +}; + +static_assert(COUNTERS_CLIENT_ID < 0 && THREAD_TRACE_CLIENT_ID < 0 && PC_SAMPLING_CLIENT_ID < 0 && + SPM_CLIENT_ID < 0, + "queue hook tags must stay disjoint from the positive ids handed out by " + "QueueController::add_callback"); +} // namespace queue_hooks +} // namespace hsa +} // namespace rocprofiler diff --git a/projects/rocprofiler-sdk/tests/CMakeLists.txt b/projects/rocprofiler-sdk/tests/CMakeLists.txt index d545a2fc9421..1f6f76a7a701 100644 --- a/projects/rocprofiler-sdk/tests/CMakeLists.txt +++ b/projects/rocprofiler-sdk/tests/CMakeLists.txt @@ -93,6 +93,7 @@ add_subdirectory(pc_sampling) add_subdirectory(hip-graph-tracing) add_subdirectory(hip-graph-forkjoin-deadlock) add_subdirectory(counter-collection) +add_subdirectory(counter-collection-per-agent) add_subdirectory(kernel-replay-concurrency) add_subdirectory(kernel-replay-local-context) add_subdirectory(perf-common) diff --git a/projects/rocprofiler-sdk/tests/counter-collection-per-agent/CMakeLists.txt b/projects/rocprofiler-sdk/tests/counter-collection-per-agent/CMakeLists.txt new file mode 100644 index 000000000000..84130df233ac --- /dev/null +++ b/projects/rocprofiler-sdk/tests/counter-collection-per-agent/CMakeLists.txt @@ -0,0 +1,52 @@ +# +# +# +cmake_minimum_required(VERSION 3.21.0 FATAL_ERROR) + +if(NOT CMAKE_HIP_COMPILER) + find_program( + amdclangpp_EXECUTABLE + NAMES amdclang++ + HINTS ${ROCM_PATH} ENV ROCM_PATH /opt/rocm + PATHS ${ROCM_PATH} ENV ROCM_PATH /opt/rocm + PATH_SUFFIXES bin llvm/bin NO_CACHE) + mark_as_advanced(amdclangpp_EXECUTABLE) + + if(amdclangpp_EXECUTABLE) + set(CMAKE_HIP_COMPILER "${amdclangpp_EXECUTABLE}") + endif() +endif() + +project( + rocprofiler-sdk-tests-counter-collection-per-agent + LANGUAGES CXX HIP + VERSION 0.0.0) + +foreach(_TYPE DEBUG MINSIZEREL RELEASE RELWITHDEBINFO) + if("${CMAKE_HIP_FLAGS_${_TYPE}}" STREQUAL "") + set(CMAKE_HIP_FLAGS_${_TYPE} "${CMAKE_CXX_FLAGS_${_TYPE}}") + endif() +endforeach() + +set(CMAKE_HIP_STANDARD 17) +set(CMAKE_HIP_EXTENSIONS OFF) +set(CMAKE_HIP_STANDARD_REQUIRED ON) + +find_package(rocprofiler-sdk REQUIRED) +find_package(Threads REQUIRED) + +# The tool and the workload live in one executable so the test can start and stop contexts +# around the measured region instead of profiling the whole process. +set_source_files_properties(per_agent_scoping_app.cpp PROPERTIES LANGUAGE HIP) +add_executable(counter-collection-per-agent-app) +target_sources(counter-collection-per-agent-app PRIVATE per_agent_scoping_app.cpp) +target_link_libraries(counter-collection-per-agent-app + PRIVATE rocprofiler-sdk::rocprofiler-sdk Threads::Threads) + +# Reports SKIP and exits 0 on a machine with fewer than two GPUs. +rocprofiler_add_integration_execute_test( + counter-collection-per-agent + TARGET counter-collection-per-agent-app + TIMEOUT 120 + LABELS "integration-tests" + SKIP_REGULAR_EXPRESSION "SKIP: per-agent scoping requires at least 2 GPUs") diff --git a/projects/rocprofiler-sdk/tests/counter-collection-per-agent/per_agent_scoping_app.cpp b/projects/rocprofiler-sdk/tests/counter-collection-per-agent/per_agent_scoping_app.cpp new file mode 100644 index 000000000000..65b96437316c --- /dev/null +++ b/projects/rocprofiler-sdk/tests/counter-collection-per-agent/per_agent_scoping_app.cpp @@ -0,0 +1,498 @@ +// MIT License +// +// Copyright (c) 2026 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. + +// End-to-end check that a dispatch counter collection context restricted to one GPU leaves +// the other GPUs alone -- neither instrumented nor serialized. +// +// The unit tests in source/lib/rocprofiler-sdk/counters/tests/per_agent_scoping_test.cpp cover +// the internal state directly. This test covers the property a user actually cares about: a +// concurrent multi-stream workload on a GPU outside the context's agent set must keep running +// concurrently. Serialization forces one dispatch at a time per agent, so the wall time of the +// workload is a direct proxy for whether the agent is serialized. +// +// Absolute times are meaningless across machines, so every timing assertion is a ratio against +// a baseline measured in the same process. The unrestricted context doubles as a positive +// control: if it does not visibly serialize the same workload, this machine cannot demonstrate +// stream concurrency at all and the timing assertions are skipped rather than guessed at. The +// record and callback assertions are exact and always run. + +#include +#include + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#define ROCPROFILER_CALL(result, msg) \ + { \ + rocprofiler_status_t CHECKSTATUS = result; \ + if(CHECKSTATUS != ROCPROFILER_STATUS_SUCCESS) \ + { \ + fprintf(stderr, \ + "[%s:%d] %s failed: %s\n", \ + __FILE__, \ + __LINE__, \ + msg, \ + rocprofiler_get_status_string(CHECKSTATUS)); \ + exit(1); \ + } \ + } + +#define HIP_CALL(cmd) \ + { \ + hipError_t e = cmd; \ + if(e != hipSuccess) \ + { \ + fprintf(stderr, "[%s:%d] HIP error: %s\n", __FILE__, __LINE__, hipGetErrorString(e)); \ + exit(1); \ + } \ + } + +namespace +{ +// Contexts are created during tool_init and started/stopped from main(). +rocprofiler_context_id_t ctx_scoped_gpu0 = {.handle = 0}; +rocprofiler_context_id_t ctx_scoped_gpu1 = {.handle = 0}; +rocprofiler_context_id_t ctx_all = {.handle = 0}; +rocprofiler_context_id_t ctx_all_second = {.handle = 0}; + +// gpu_agents[i] is the rocprofiler agent for HIP device i. Populated in tool_init by +// matching each HIP device's PCI domain/BDF to the rocprofiler agent with the same +// coordinates so that the contexts and assertions both refer to the same physical device +// regardless of the order rocprofiler enumerates agents or how visible-device variables +// reorder HIP ordinals. +std::vector gpu_agents = {}; + +// HIP ordinal that could not be matched to a rocprofiler agent, or -1 when every device +// mapped. Reported by main() so a skip is distinguishable from a machine with one GPU. +int unmapped_device = -1; + +// Intermediate storage used to build the HIP-to-rocprofiler mapping: keeps the +// location_id (BDF) and PCI domain alongside the agent id until the mapping is done. +struct agent_info_t +{ + rocprofiler_agent_id_t id; + uint32_t location_id; // (bus<<8)|(device<<3)|function + uint32_t domain; +}; +std::vector all_gpu_agent_info = {}; + +std::mutex observed_mutex = {}; +std::map dispatch_per_agent = {}; +std::map records_per_agent = {}; + +void +reset_observations() +{ + auto lk = std::unique_lock{observed_mutex}; + dispatch_per_agent.clear(); + records_per_agent.clear(); +} + +size_t +dispatches_on(rocprofiler_agent_id_t agent) +{ + auto lk = std::unique_lock{observed_mutex}; + auto itr = dispatch_per_agent.find(agent.handle); + return (itr == dispatch_per_agent.end()) ? 0 : itr->second; +} + +size_t +records_on(rocprofiler_agent_id_t agent) +{ + auto lk = std::unique_lock{observed_mutex}; + auto itr = records_per_agent.find(agent.handle); + return (itr == records_per_agent.end()) ? 0 : itr->second; +} + +void +record_callback(rocprofiler_dispatch_counting_service_data_t dispatch_data, + rocprofiler_counter_record_t*, + size_t record_count, + rocprofiler_user_data_t, + void*) +{ + if(record_count == 0) return; + auto lk = std::unique_lock{observed_mutex}; + records_per_agent[dispatch_data.dispatch_info.agent_id.handle]++; +} + +void +dispatch_callback(rocprofiler_dispatch_counting_service_data_t dispatch_data, + rocprofiler_counter_config_id_t* config, + rocprofiler_user_data_t*, + void*) +{ + const auto agent_id = dispatch_data.dispatch_info.agent_id; + + { + auto lk = std::unique_lock{observed_mutex}; + dispatch_per_agent[agent_id.handle]++; + } + + static std::shared_mutex cache_mutex = {}; + static std::unordered_map cache = {}; + + auto search = [&]() { + if(auto pos = cache.find(agent_id.handle); pos != cache.end()) + { + *config = pos->second; + return true; + } + return false; + }; + + { + auto rlock = std::shared_lock{cache_mutex}; + if(search()) return; + } + + auto wlock = std::unique_lock{cache_mutex}; + if(search()) return; + + auto supported = std::vector{}; + ROCPROFILER_CALL(rocprofiler_iterate_agent_supported_counters( + agent_id, + [](rocprofiler_agent_id_t, + rocprofiler_counter_id_t* counters, + size_t num_counters, + void* user_data) { + auto* vec = + static_cast*>(user_data); + for(size_t i = 0; i < num_counters; i++) + vec->push_back(counters[i]); + return ROCPROFILER_STATUS_SUCCESS; + }, + static_cast(&supported)), + "iterate supported counters"); + + auto collect = std::vector{}; + for(auto& counter : supported) + { + rocprofiler_counter_info_v0_t info; + ROCPROFILER_CALL(rocprofiler_query_counter_info( + counter, ROCPROFILER_COUNTER_INFO_VERSION_0, (void*) &info), + "query counter info"); + if(strcmp(info.name, "SQ_WAVES") == 0) collect.push_back(counter); + } + + if(collect.empty()) return; + + rocprofiler_counter_config_id_t profile = {.handle = 0}; + ROCPROFILER_CALL( + rocprofiler_create_counter_config(agent_id, collect.data(), collect.size(), &profile), + "create counter config"); + + cache.emplace(agent_id.handle, profile); + *config = profile; +} + +void +configure_cc(rocprofiler_context_id_t* ctx, const rocprofiler_agent_id_t* agent) +{ + ROCPROFILER_CALL(rocprofiler_create_context(ctx), "create context"); + ROCPROFILER_CALL(rocprofiler_configure_callback_dispatch_counting_service( + *ctx, dispatch_callback, nullptr, record_callback, nullptr), + "configure dispatch counting service"); + if(agent) + ROCPROFILER_CALL(rocprofiler_dispatch_counting_service_set_agents(*ctx, agent, 1), + "restrict counting service to agent"); +} + +rocprofiler_status_t +collect_gpu_agents(rocprofiler_agent_version_t, const void** agents, size_t num_agents, void*) +{ + for(size_t i = 0; i < num_agents; ++i) + { + const auto* agent = static_cast(agents[i]); + if(agent->type == ROCPROFILER_AGENT_TYPE_GPU) + all_gpu_agent_info.push_back({agent->id, agent->location_id, agent->domain}); + } + return ROCPROFILER_STATUS_SUCCESS; +} + +int +tool_init(rocprofiler_client_finalize_t, void*) +{ + ROCPROFILER_CALL(rocprofiler_query_available_agents(ROCPROFILER_AGENT_INFO_VERSION_0, + collect_gpu_agents, + sizeof(rocprofiler_agent_v0_t), + nullptr), + "query available agents"); + + // Nothing to configure when the machine cannot express the scenario; main() reports the + // skip so the reason is visible in the test output. + if(all_gpu_agent_info.size() < 2) return 0; + + // Map each HIP device ordinal to its rocprofiler agent using PCI domain/BDF so that + // gpu_agents[i] is the agent for HIP device i, regardless of the order rocprofiler + // enumerates agents or how ROCR_VISIBLE_DEVICES / HIP_VISIBLE_DEVICES reorders ordinals. + int device_count = 0; + if(hipGetDeviceCount(&device_count) != hipSuccess || device_count < 2) return 0; + + // A device with no matching agent must not be skipped silently: dropping it would shift + // every later ordinal, so gpu_agents[i] would stop being HIP device i and the contexts + // would be scoped to a different physical GPU than the workload runs on. Bail instead and + // let main() report the skip. + for(int dev = 0; dev < device_count; ++dev) + { + hipDeviceProp_t prop{}; + if(hipGetDeviceProperties(&prop, dev) != hipSuccess) return 0; + const auto bus = static_cast(prop.pciBusID); + const auto did = static_cast(prop.pciDeviceID); + const auto dom = static_cast(prop.pciDomainID); + + bool matched = false; + for(const auto& info : all_gpu_agent_info) + { + if(info.domain == dom && (info.location_id >> 8) == bus && + ((info.location_id >> 3) & 0x1Fu) == did) + { + gpu_agents.push_back(info.id); + matched = true; + break; + } + } + + // Only the first two ordinals are used, so an unmapped device beyond them is harmless. + if(!matched) + { + unmapped_device = dev; + break; + } + } + + if(gpu_agents.size() < 2) return 0; + + configure_cc(&ctx_scoped_gpu0, &gpu_agents[0]); + configure_cc(&ctx_scoped_gpu1, &gpu_agents[1]); + configure_cc(&ctx_all, nullptr); + configure_cc(&ctx_all_second, nullptr); + + return 0; +} + +void +tool_fini(void*) +{} +} // namespace + +extern "C" rocprofiler_tool_configure_result_t* +rocprofiler_configure(uint32_t, const char*, uint32_t, rocprofiler_client_id_t* id) +{ + id->name = "PerAgentScopingTest"; + static auto cfg = rocprofiler_tool_configure_result_t{ + sizeof(rocprofiler_tool_configure_result_t), &tool_init, &tool_fini, nullptr}; + return &cfg; +} + +// ------------------------------------------------------------------------------------------- +// Workload +// ------------------------------------------------------------------------------------------- + +__global__ void +spin_kernel(long long cycles, float* out) +{ + long long start = clock64(); + float acc = 0.0F; + while(clock64() - start < cycles) + acc += 1.0F; + // Never true for a 64-thread block; keeps the loop from being optimized away without + // adding a real store to the timed path. + if(threadIdx.x == 1024) *out = acc; +} + +namespace +{ +constexpr int kNumStreams = 8; +constexpr int kKernelsPer = 4; +constexpr long kSpinCycles = 2000000; + +// Launches kKernelsPer kernels on each of kNumStreams streams and waits for all of them. When +// the agent is not serialized the streams overlap and the wall time is roughly kKernelsPer +// kernels long; when it is serialized every dispatch runs alone and the wall time is roughly +// kNumStreams * kKernelsPer kernels long. +double +time_concurrent_workload(int device) +{ + HIP_CALL(hipSetDevice(device)); + + hipStream_t streams[kNumStreams]; + for(auto& s : streams) + HIP_CALL(hipStreamCreate(&s)); + + float* out = nullptr; + HIP_CALL(hipMalloc(&out, sizeof(float))); + + // Warm up so queue creation and code object loading are not inside the measurement. + for(auto& s : streams) + hipLaunchKernelGGL(spin_kernel, dim3(1), dim3(64), 0, s, kSpinCycles / 10, out); + HIP_CALL(hipDeviceSynchronize()); + + auto t0 = std::chrono::steady_clock::now(); + for(int k = 0; k < kKernelsPer; ++k) + for(auto& s : streams) + hipLaunchKernelGGL(spin_kernel, dim3(1), dim3(64), 0, s, kSpinCycles, out); + HIP_CALL(hipDeviceSynchronize()); + auto t1 = std::chrono::steady_clock::now(); + + for(auto& s : streams) + HIP_CALL(hipStreamDestroy(s)); + HIP_CALL(hipFree(out)); + + return std::chrono::duration(t1 - t0).count(); +} + +int failures = 0; + +void +check(bool cond, const std::string& what) +{ + printf("[%s] %s\n", cond ? "PASS" : "FAIL", what.c_str()); + if(!cond) ++failures; +} +} // namespace + +int +main() +{ + int device_count = 0; + HIP_CALL(hipGetDeviceCount(&device_count)); + + if(device_count < 2 || gpu_agents.size() < 2) + { + if(unmapped_device >= 0 && unmapped_device < 2) + printf("SKIP: HIP device %d has no rocprofiler agent with matching PCI " + "coordinates, so per-agent scoping cannot be checked in this configuration\n", + unmapped_device); + else + printf("SKIP: per-agent scoping requires at least 2 GPUs (found %d)\n", device_count); + return 0; + } + + const auto agent_0 = gpu_agents[0]; + const auto agent_1 = gpu_agents[1]; + + // Phase 1: no counter collection anywhere. Reference for an unserialized GPU-0. + const double baseline_ms = time_concurrent_workload(0); + + // Phase 2: unrestricted context. This is what every tool got before per-agent scoping, and + // it doubles as the positive control for the timing measurement. + reset_observations(); + ROCPROFILER_CALL(rocprofiler_start_context(ctx_all), "start unrestricted context"); + const double unrestricted_ms = time_concurrent_workload(0); + ROCPROFILER_CALL(rocprofiler_stop_context(ctx_all), "stop unrestricted context"); + const size_t unrestricted_dispatches_gpu0 = dispatches_on(agent_0); + + // Phase 3: context restricted to GPU-1. GPU-1 must still be collected on; GPU-0 must be + // untouched. + reset_observations(); + ROCPROFILER_CALL(rocprofiler_start_context(ctx_scoped_gpu1), "start scoped context"); + time_concurrent_workload(1); + const double scoped_ms = time_concurrent_workload(0); + ROCPROFILER_CALL(rocprofiler_stop_context(ctx_scoped_gpu1), "stop scoped context"); + const size_t scoped_records_gpu1 = records_on(agent_1); + const size_t scoped_dispatches_gpu0 = dispatches_on(agent_0); + const size_t scoped_records_gpu0 = records_on(agent_0); + + printf("baseline=%.2fms unrestricted=%.2fms scoped=%.2fms (%d streams x %d kernels)\n", + baseline_ms, + unrestricted_ms, + scoped_ms, + kNumStreams, + kKernelsPer); + // This test only runs on a multi-GPU machine, so print the mapping and the raw counts the + // assertions below read: a control failure is otherwise indistinguishable from a scoping bug. + printf("hip device 0 -> agent %lu, hip device 1 -> agent %lu\n", + static_cast(agent_0.handle), + static_cast(agent_1.handle)); + printf("unrestricted: dispatches(gpu0)=%zu | scoped: records(gpu1)=%zu dispatches(gpu0)=%zu " + "records(gpu0)=%zu\n", + unrestricted_dispatches_gpu0, + scoped_records_gpu1, + scoped_dispatches_gpu0, + scoped_records_gpu0); + + // Exact assertions. These hold regardless of how fast the machine is. + check(unrestricted_dispatches_gpu0 > 0, + "unrestricted context sees dispatches on GPU-0 (control)"); + check(scoped_records_gpu1 > 0, "scoped context still collects counters on its own agent"); + check(scoped_dispatches_gpu0 == 0, + "scoped context never receives a dispatch callback for GPU-0"); + check(scoped_records_gpu0 == 0, "scoped context produces no records for GPU-0"); + + // Timing assertions, but only when the positive control shows this machine can actually + // demonstrate the difference. Otherwise the workload never overlapped to begin with and a + // ratio would be measuring noise. + const double control_ratio = unrestricted_ms / baseline_ms; + if(control_ratio < 2.0) + { + printf("SKIP timing checks: unrestricted context only slowed GPU-0 by %.2fx, so this " + "machine does not show stream concurrency clearly enough to measure\n", + control_ratio); + } + else + { + const double scoped_ratio = scoped_ms / baseline_ms; + printf("serialization slowdown: unrestricted=%.2fx scoped=%.2fx\n", + control_ratio, + scoped_ratio); + check(scoped_ratio < 2.0, + "a context scoped to GPU-1 leaves GPU-0 running at baseline speed"); + } + + // Conflict semantics: disjoint agent sets may run together, overlapping ones may not. + { + auto status_a = rocprofiler_start_context(ctx_scoped_gpu0); + auto status_b = rocprofiler_start_context(ctx_scoped_gpu1); + check(status_a == ROCPROFILER_STATUS_SUCCESS && status_b == ROCPROFILER_STATUS_SUCCESS, + "two counter collection contexts on disjoint agents can both be active"); + ROCPROFILER_CALL(rocprofiler_stop_context(ctx_scoped_gpu1), "stop gpu1 context"); + ROCPROFILER_CALL(rocprofiler_stop_context(ctx_scoped_gpu0), "stop gpu0 context"); + } + + { + ROCPROFILER_CALL(rocprofiler_start_context(ctx_all), "start unrestricted context"); + auto status = rocprofiler_start_context(ctx_all_second); + check(status == ROCPROFILER_STATUS_ERROR_CONTEXT_CONFLICT, + "two unrestricted counter collection contexts still conflict"); + ROCPROFILER_CALL(rocprofiler_stop_context(ctx_all), "stop unrestricted context"); + } + + if(failures > 0) + { + printf("%d check(s) failed\n", failures); + return 1; + } + + printf("all per-agent scoping checks passed\n"); + return 0; +}