From de1ce236e473c415f27f37c6acb1627c3aa0f47a Mon Sep 17 00:00:00 2001 From: Adeem Jassani Date: Sun, 29 Mar 2026 17:43:24 +0000 Subject: [PATCH 1/8] feat: add inter-stream dependency tracking for rocprofiler-sdk backend Capture hipEventRecord and hipStreamWaitEvent calls during profiling to expose inter-stream synchronization dependencies in Chrome/Perfetto traces. This data enables downstream tools to reconstruct the communication-compute overlap and causal ordering between GPU streams in multi-stream workloads (e.g. DDP, pipeline parallelism). New activity types: - rocprofEventRecordRow: records which event was placed on which stream - rocprofSyncRow: records sync operations with source stream/event info The metadata is emitted as JSON fields (hip_event, hip_stream, sync_type, wait_on_stream, wait_on_hip_event_record_corr_id) that can be parsed by trace analysis tools to build a dependency graph. hipEventRecord is removed from the callback omit list so that event record calls are captured by the profiler. Test: InterStreamDependencyTest verifies that event record and sync activities are emitted with correct dependency metadata in the Chrome trace JSON output. --- libkineto/src/RocLogger.h | 54 +++++++ libkineto/src/RocmActivityProfiler.cpp | 8 + libkineto/src/RocprofActivity_inl.h | 57 +++++++ libkineto/src/RocprofLogger.cpp | 158 ++++++++++++++++++++ libkineto/test/RocmActivityProfilerTest.cpp | 135 +++++++++++++++++ 5 files changed, 412 insertions(+) diff --git a/libkineto/src/RocLogger.h b/libkineto/src/RocLogger.h index a59ffada5..61f23599e 100644 --- a/libkineto/src/RocLogger.h +++ b/libkineto/src/RocLogger.h @@ -69,6 +69,8 @@ typedef enum { ROCTRACER_ACTIVITY_COPY, ROCTRACER_ACTIVITY_MALLOC, ROCTRACER_ACTIVITY_ASYNC, + ROCTRACER_ACTIVITY_EVENT_RECORD, + ROCTRACER_ACTIVITY_SYNC, ROCTRACER_ACTIVITY_NONE } rocprof_activity_types; @@ -213,3 +215,55 @@ struct rocprofAsyncRow : public rocprofBase { uint64_t queue; std::string kernelName; }; + +enum rocprofSyncType { + ROCPROF_SYNC_STREAM_WAIT_EVENT = 0, + ROCPROF_SYNC_EVENT_SYNCHRONIZE, + ROCPROF_SYNC_STREAM_SYNCHRONIZE, + ROCPROF_SYNC_DEVICE_SYNCHRONIZE, +}; + +struct rocprofEventRecordRow : public rocprofRow { + rocprofEventRecordRow(uint64_t id, + uint32_t domain, + uint32_t cid, + uint32_t pid, + uint32_t tid, + uint64_t begin, + uint64_t end, + hipEvent_t event, + hipStream_t stream) + : rocprofRow(id, domain, cid, pid, tid, begin, end, + ROCTRACER_ACTIVITY_EVENT_RECORD), + event(event), + stream(stream) {} + hipEvent_t event; + hipStream_t stream; +}; + +struct rocprofSyncRow : public rocprofRow { + rocprofSyncRow(uint64_t id, + uint32_t domain, + uint32_t cid, + uint32_t pid, + uint32_t tid, + uint64_t begin, + uint64_t end, + rocprofSyncType syncType, + hipStream_t stream, + hipEvent_t event, + hipStream_t srcStream, + uint64_t srcCorrId) + : rocprofRow(id, domain, cid, pid, tid, begin, end, + ROCTRACER_ACTIVITY_SYNC), + syncType(syncType), + stream(stream), + event(event), + srcStream(srcStream), + srcCorrId(srcCorrId) {} + rocprofSyncType syncType; + hipStream_t stream; + hipEvent_t event; + hipStream_t srcStream; + uint64_t srcCorrId; +}; diff --git a/libkineto/src/RocmActivityProfiler.cpp b/libkineto/src/RocmActivityProfiler.cpp index 120d5b380..9999013ac 100644 --- a/libkineto/src/RocmActivityProfiler.cpp +++ b/libkineto/src/RocmActivityProfiler.cpp @@ -273,6 +273,14 @@ void RocmActivityProfiler::handleRocprofActivity( handleGpuActivity( reinterpret_cast(record), logger); break; + case ROCTRACER_ACTIVITY_EVENT_RECORD: + handleRuntimeActivity( + reinterpret_cast(record), logger); + break; + case ROCTRACER_ACTIVITY_SYNC: + handleRuntimeActivity( + reinterpret_cast(record), logger); + break; case ROCTRACER_ACTIVITY_NONE: default: LOG(WARNING) << "Unexpected activity type: " << record->type; diff --git a/libkineto/src/RocprofActivity_inl.h b/libkineto/src/RocprofActivity_inl.h index bb0dcc6e4..606d3f47c 100644 --- a/libkineto/src/RocprofActivity_inl.h +++ b/libkineto/src/RocprofActivity_inl.h @@ -239,6 +239,63 @@ inline const std::string RuntimeActivity::metadataJson() const raw().ptr); } +template <> +inline const std::string RuntimeActivity::metadataJson() + const { + return fmt::format( + R"JSON( + "cid": {}, "correlation": {}, + "hip_event": "{}", "hip_stream": "{}")JSON", + raw().cid, + raw().id, + fmt::ptr(raw().event), + fmt::ptr(raw().stream)); +} + +template <> +inline const std::string RuntimeActivity::metadataJson() const { + static const char* syncTypeNames[] = { + "stream_wait_event", + "event_synchronize", + "stream_synchronize", + "device_synchronize", + }; + const char* syncName = (raw().syncType >= 0 && raw().syncType <= 3) + ? syncTypeNames[raw().syncType] + : "unknown"; + + std::string meta = fmt::format( + R"JSON( + "cid": {}, "correlation": {}, + "sync_type": "{}")JSON", + raw().cid, + raw().id, + syncName); + + if (raw().syncType == ROCPROF_SYNC_STREAM_WAIT_EVENT && raw().srcStream) { + meta += fmt::format( + R"JSON(, + "wait_on_stream": "{}", + "wait_on_hip_event_record_corr_id": {}, + "wait_on_hip_event": "{}")JSON", + fmt::ptr(raw().srcStream), + raw().srcCorrId, + fmt::ptr(raw().event)); + } else if (raw().stream) { + meta += fmt::format( + R"JSON(, + "hip_stream": "{}")JSON", + fmt::ptr(raw().stream)); + } + if (raw().event && raw().syncType == ROCPROF_SYNC_EVENT_SYNCHRONIZE) { + meta += fmt::format( + R"JSON(, + "hip_event": "{}")JSON", + fmt::ptr(raw().event)); + } + return meta; +} + template inline const std::string RuntimeActivity::metadataJson() const { return fmt::format( diff --git a/libkineto/src/RocprofLogger.cpp b/libkineto/src/RocprofLogger.cpp index 24a8b34bf..49ad52c5e 100644 --- a/libkineto/src/RocprofLogger.cpp +++ b/libkineto/src/RocprofLogger.cpp @@ -250,6 +250,89 @@ bool isMallocApi(uint32_t id) { return false; } +bool isEventRecordApi(uint32_t id) { + switch (id) { + case ROCPROFILER_HIP_RUNTIME_API_ID_hipEventRecord: + case ROCPROFILER_HIP_RUNTIME_API_ID_hipEventRecord_spt: +#ifdef ROCPROFILER_HIP_RUNTIME_API_ID_hipEventRecordWithFlags + case ROCPROFILER_HIP_RUNTIME_API_ID_hipEventRecordWithFlags: +#endif + return true; + break; + default:; + } + return false; +} + +bool isSyncApi(uint32_t id) { + switch (id) { + case ROCPROFILER_HIP_RUNTIME_API_ID_hipStreamWaitEvent: + case ROCPROFILER_HIP_RUNTIME_API_ID_hipStreamWaitEvent_spt: + case ROCPROFILER_HIP_RUNTIME_API_ID_hipEventSynchronize: + case ROCPROFILER_HIP_RUNTIME_API_ID_hipStreamSynchronize: + case ROCPROFILER_HIP_RUNTIME_API_ID_hipStreamSynchronize_spt: + case ROCPROFILER_HIP_RUNTIME_API_ID_hipDeviceSynchronize: + return true; + break; + default:; + } + return false; +} + +struct event_record_args { + hipEvent_t event{nullptr}; + hipStream_t stream{nullptr}; +}; +auto extract_event_record_args = + []([[maybe_unused]] rocprofiler_callback_tracing_kind_t kind, + [[maybe_unused]] rocprofiler_tracing_operation_t operation, + [[maybe_unused]] uint32_t arg_num, + const void* const arg_value_addr, + [[maybe_unused]] int32_t indirection_count, + [[maybe_unused]] const char* arg_type, + const char* arg_name, + [[maybe_unused]] const char* arg_value_str, + [[maybe_unused]] int32_t dereference_count, + void* cb_data) -> int { + auto& args = *(static_cast(cb_data)); + if (strcmp("event", arg_name) == 0) + args.event = *(reinterpret_cast(arg_value_addr)); + else if (strcmp("stream", arg_name) == 0) + args.stream = *(reinterpret_cast(arg_value_addr)); + return 0; +}; + +struct sync_args { + hipStream_t stream{nullptr}; + hipEvent_t event{nullptr}; +}; +auto extract_sync_args = + []([[maybe_unused]] rocprofiler_callback_tracing_kind_t kind, + [[maybe_unused]] rocprofiler_tracing_operation_t operation, + [[maybe_unused]] uint32_t arg_num, + const void* const arg_value_addr, + [[maybe_unused]] int32_t indirection_count, + [[maybe_unused]] const char* arg_type, + const char* arg_name, + [[maybe_unused]] const char* arg_value_str, + [[maybe_unused]] int32_t dereference_count, + void* cb_data) -> int { + auto& args = *(static_cast(cb_data)); + if (strcmp("stream", arg_name) == 0) + args.stream = *(reinterpret_cast(arg_value_addr)); + else if (strcmp("event", arg_name) == 0) + args.event = *(reinterpret_cast(arg_value_addr)); + return 0; +}; + +// Maps hipEvent_t -> {hipStream_t, correlation_id} from hipEventRecord calls +struct EventMapEntry { + hipStream_t stream{nullptr}; + uint64_t corrId{0}; +}; +std::mutex g_eventMapMutex; +std::unordered_map g_eventMap; + class RocprofApiIdList : public ApiIdList { public: RocprofApiIdList(callback_name_info& names); @@ -672,6 +755,81 @@ void RocprofLogger::api_callback( args.size); insert_row_to_buffer(row); } + // Event Record + else if (isEventRecordApi(record.operation)) { + event_record_args args; + rocprofiler_iterate_callback_tracing_kind_operation_args( + record, extract_event_record_args, 1, &args); + + { + std::lock_guard lock(g_eventMapMutex); + g_eventMap[static_cast(args.event)] = + {args.stream, record.correlation_id.internal}; + } + + rocprofEventRecordRow* row = new rocprofEventRecordRow( + record.correlation_id.internal, + record.kind, + record.operation, + processId(), + systemThreadId(), + startTime, + endTime, + args.event, + args.stream); + insert_row_to_buffer(row); + } + // Sync APIs (stream wait event, event sync, stream sync, device sync) + else if (isSyncApi(record.operation)) { + sync_args args; + rocprofiler_iterate_callback_tracing_kind_operation_args( + record, extract_sync_args, 1, &args); + + rocprofSyncType syncType; + hipStream_t srcStream = nullptr; + uint64_t srcCorrId = 0; + + switch (record.operation) { + case ROCPROFILER_HIP_RUNTIME_API_ID_hipStreamWaitEvent: + case ROCPROFILER_HIP_RUNTIME_API_ID_hipStreamWaitEvent_spt: + { + syncType = ROCPROF_SYNC_STREAM_WAIT_EVENT; + std::lock_guard lock(g_eventMapMutex); + auto it = g_eventMap.find(static_cast(args.event)); + if (it != g_eventMap.end()) { + srcStream = it->second.stream; + srcCorrId = it->second.corrId; + } + break; + } + case ROCPROFILER_HIP_RUNTIME_API_ID_hipEventSynchronize: + syncType = ROCPROF_SYNC_EVENT_SYNCHRONIZE; + break; + case ROCPROFILER_HIP_RUNTIME_API_ID_hipStreamSynchronize: + case ROCPROFILER_HIP_RUNTIME_API_ID_hipStreamSynchronize_spt: + syncType = ROCPROF_SYNC_STREAM_SYNCHRONIZE; + break; + case ROCPROFILER_HIP_RUNTIME_API_ID_hipDeviceSynchronize: + default: + syncType = ROCPROF_SYNC_DEVICE_SYNCHRONIZE; + break; + } + + rocprofSyncRow* row = new rocprofSyncRow( + record.correlation_id.internal, + record.kind, + record.operation, + processId(), + systemThreadId(), + startTime, + endTime, + syncType, + args.stream, + args.event, + srcStream, + srcCorrId); + insert_row_to_buffer(row); + } // Default Records else { struct { hipStream_t stream{nullptr}; } default_args; diff --git a/libkineto/test/RocmActivityProfilerTest.cpp b/libkineto/test/RocmActivityProfilerTest.cpp index c8a3975dd..85d735ade 100644 --- a/libkineto/test/RocmActivityProfilerTest.cpp +++ b/libkineto/test/RocmActivityProfilerTest.cpp @@ -310,6 +310,52 @@ struct MockRocLogger { activities_.push_back(row); } +#ifndef ROCTRACER_FALLBACK + void addEventRecordActivity( + int64_t start_ns, + int64_t end_ns, + int64_t correlation, + hipEvent_t event, + hipStream_t stream) { + rocprofEventRecordRow* row = new rocprofEventRecordRow( + correlation, + RUNTIME_DOMAIN, + ROCPROFILER_HIP_RUNTIME_API_ID_hipEventRecord, + processId(), + systemThreadId(), + start_ns, + end_ns, + event, + stream); + activities_.push_back(row); + } + + void addSyncActivity( + int64_t start_ns, + int64_t end_ns, + int64_t correlation, + rocprofSyncType syncType, + hipStream_t stream, + hipEvent_t event, + hipStream_t srcStream, + uint64_t srcCorrId) { + rocprofSyncRow* row = new rocprofSyncRow( + correlation, + RUNTIME_DOMAIN, + ROCPROFILER_HIP_RUNTIME_API_ID_hipStreamWaitEvent, + processId(), + systemThreadId(), + start_ns, + end_ns, + syncType, + stream, + event, + srcStream, + srcCorrId); + activities_.push_back(row); + } +#endif + ~MockRocLogger() { while (!activities_.empty()) { auto act = activities_.back(); @@ -1044,3 +1090,92 @@ TEST_F(RocmActivityProfilerTest, JsonGPUIDSortTest) { } #endif } + +#ifndef ROCTRACER_FALLBACK +TEST_F(RocmActivityProfilerTest, InterStreamDependencyTest) { + std::vector log_modules({"RocmActivityProfiler.cpp"}); + SET_LOG_VERBOSITY_LEVEL(2, log_modules); + + RocmActivityProfiler profiler(rocActivities_, /*cpu only*/ false); + int64_t start_time_ns = + libkineto::timeSinceEpoch(std::chrono::system_clock::now()); + int64_t duration_ns = 300; + auto start_time = time_point(nanoseconds(start_time_ns)); + profiler.configure(*cfg_, start_time); + profiler.startTrace(start_time); + profiler.stopTrace(start_time + nanoseconds(duration_ns)); + profiler.recordThreadInfo(); + + auto cpuOps = std::make_unique( + start_time_ns, start_time_ns + duration_ns); + cpuOps->addOp("op1", start_time_ns + 10, start_time_ns + 30, 1); + profiler.transferCpuTrace(std::move(cpuOps)); + + // Simulate: hipEventRecord(event=0xA, stream=0x1) with corr=10 + // then hipStreamWaitEvent(stream=0x2, event=0xA) with corr=20 + auto gpuOps = std::make_unique(); + hipEvent_t fakeEvent = reinterpret_cast(0xA); + hipStream_t stream0 = reinterpret_cast(0x1); + hipStream_t stream1 = reinterpret_cast(0x2); + + gpuOps->addEventRecordActivity( + start_time_ns + 20, start_time_ns + 25, 10, fakeEvent, stream0); + gpuOps->addSyncActivity( + start_time_ns + 30, start_time_ns + 35, 20, + ROCPROF_SYNC_STREAM_WAIT_EVENT, stream1, fakeEvent, stream0, 10); + gpuOps->addKernelActivity(start_time_ns + 50, start_time_ns + 100, 1); + rocActivities_.activityLogger = std::move(gpuOps); + + auto logger = std::make_unique(*cfg_); + profiler.processTrace(*logger); + profiler.reset(); + + ActivityTrace trace(std::move(logger), loggerFactory); + + // Verify event record and sync activities are present + int eventRecordCount = 0; + int syncCount = 0; + for (auto& activity : *trace.activities()) { + if (activity->name() == "hipEventRecord") { + eventRecordCount++; + } + if (activity->name() == "hipStreamWaitEvent") { + syncCount++; + } + } + EXPECT_EQ(eventRecordCount, 1); + EXPECT_EQ(syncCount, 1); + + // Verify JSON output contains dependency metadata + char filename[] = "/tmp/libkineto_interstream_testXXXXXX.json"; + { int tmp_fd = mkstemps(filename, 5); + if (tmp_fd >= 0) close(tmp_fd); } + trace.save(filename); + + std::ifstream f(filename); + nlohmann::json j = nlohmann::json::parse(f); + auto& traceEvents = j["traceEvents"]; + + bool foundEventRecord = false; + bool foundSyncWithDep = false; + for (auto& ev : traceEvents) { + if (ev.value("name", "") == "hipEventRecord") { + foundEventRecord = true; + EXPECT_TRUE(ev["args"].contains("hip_event")); + EXPECT_TRUE(ev["args"].contains("hip_stream")); + } + if (ev.value("name", "") == "hipStreamWaitEvent") { + foundSyncWithDep = true; + EXPECT_TRUE(ev["args"].contains("sync_type")); + EXPECT_EQ(ev["args"]["sync_type"], "stream_wait_event"); + EXPECT_TRUE(ev["args"].contains("wait_on_stream")); + EXPECT_TRUE(ev["args"].contains("wait_on_hip_event_record_corr_id")); + EXPECT_EQ(ev["args"]["wait_on_hip_event_record_corr_id"], 10); + } + } + EXPECT_TRUE(foundEventRecord); + EXPECT_TRUE(foundSyncWithDep); + + unlink(filename); +} +#endif From 680ab1647e094a2c2d064c2581b5b0aa687d8b23 Mon Sep 17 00:00:00 2001 From: Adeem Jassani Date: Fri, 10 Apr 2026 17:06:19 +0000 Subject: [PATCH 2/8] fix: always emit hip_stream and hip_event in sync metadata JSON Bug A: unresolved hipStreamWaitEvent events dropped the event handle entirely, making it impossible to reconstruct the dependency. Bug B: resolved events emitted wait_on_stream (producer) but lost hip_stream (consumer). Now all sync events always emit hip_stream (consumer) and hip_event when available, plus the wait_on_* fields when the lookup succeeds. --- libkineto/src/RocprofActivity_inl.h | 21 ++++++++++----------- 1 file changed, 10 insertions(+), 11 deletions(-) diff --git a/libkineto/src/RocprofActivity_inl.h b/libkineto/src/RocprofActivity_inl.h index 606d3f47c..a7998c992 100644 --- a/libkineto/src/RocprofActivity_inl.h +++ b/libkineto/src/RocprofActivity_inl.h @@ -272,27 +272,26 @@ inline const std::string RuntimeActivity::metadataJson() const { raw().id, syncName); - if (raw().syncType == ROCPROF_SYNC_STREAM_WAIT_EVENT && raw().srcStream) { - meta += fmt::format( - R"JSON(, - "wait_on_stream": "{}", - "wait_on_hip_event_record_corr_id": {}, - "wait_on_hip_event": "{}")JSON", - fmt::ptr(raw().srcStream), - raw().srcCorrId, - fmt::ptr(raw().event)); - } else if (raw().stream) { + if (raw().stream) { meta += fmt::format( R"JSON(, "hip_stream": "{}")JSON", fmt::ptr(raw().stream)); } - if (raw().event && raw().syncType == ROCPROF_SYNC_EVENT_SYNCHRONIZE) { + if (raw().event) { meta += fmt::format( R"JSON(, "hip_event": "{}")JSON", fmt::ptr(raw().event)); } + if (raw().syncType == ROCPROF_SYNC_STREAM_WAIT_EVENT && raw().srcStream) { + meta += fmt::format( + R"JSON(, + "wait_on_stream": "{}", + "wait_on_hip_event_record_corr_id": {})JSON", + fmt::ptr(raw().srcStream), + raw().srcCorrId); + } return meta; } From 9775e2c84872afc981474fa54a8059c294b021b4 Mon Sep 17 00:00:00 2001 From: Adeem Jassani Date: Fri, 10 Apr 2026 17:28:02 +0000 Subject: [PATCH 3/8] fix: always emit hip_stream even for default stream (0x0) The consumer stream in hipStreamWaitEvent can be the default stream (null pointer), which is still meaningful for dependency resolution. Emit it unconditionally so the post-processor can always find both producer and consumer. --- libkineto/src/RocprofActivity_inl.h | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/libkineto/src/RocprofActivity_inl.h b/libkineto/src/RocprofActivity_inl.h index a7998c992..eaf56f3d2 100644 --- a/libkineto/src/RocprofActivity_inl.h +++ b/libkineto/src/RocprofActivity_inl.h @@ -272,12 +272,10 @@ inline const std::string RuntimeActivity::metadataJson() const { raw().id, syncName); - if (raw().stream) { - meta += fmt::format( - R"JSON(, + meta += fmt::format( + R"JSON(, "hip_stream": "{}")JSON", - fmt::ptr(raw().stream)); - } + fmt::ptr(raw().stream)); if (raw().event) { meta += fmt::format( R"JSON(, From f2374e0369bbe0c8d7e739d53194b674cd0afbb6 Mon Sep 17 00:00:00 2001 From: Adeem Jassani Date: Fri, 10 Apr 2026 19:40:43 +0000 Subject: [PATCH 4/8] fix: use srcCorrId instead of srcStream to gate dependency serialization MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The default/compute stream in HIP is represented as nullptr (0x0). When hipEventRecord records an event on the default stream, g_eventMap stores srcStream=nullptr. The callback correctly retrieves this, but metadataJson() was checking raw().srcStream as a null-pointer guard, which silently dropped the resolution for all default-stream events. This caused 50% of hipStreamWaitEvent dependencies (the comp→comm direction in DDP) to appear unresolved in the trace despite the g_eventMap lookup succeeding at callback time. Fix: check srcCorrId (non-zero when lookup succeeded) instead of srcStream (which is legitimately nullptr for the default stream). --- libkineto/src/RocprofActivity_inl.h | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/libkineto/src/RocprofActivity_inl.h b/libkineto/src/RocprofActivity_inl.h index eaf56f3d2..dd8e4fb15 100644 --- a/libkineto/src/RocprofActivity_inl.h +++ b/libkineto/src/RocprofActivity_inl.h @@ -282,7 +282,7 @@ inline const std::string RuntimeActivity::metadataJson() const { "hip_event": "{}")JSON", fmt::ptr(raw().event)); } - if (raw().syncType == ROCPROF_SYNC_STREAM_WAIT_EVENT && raw().srcStream) { + if (raw().syncType == ROCPROF_SYNC_STREAM_WAIT_EVENT && raw().srcCorrId) { meta += fmt::format( R"JSON(, "wait_on_stream": "{}", From 13fe0177f6dff874bbc433d87a107a0a8243b117 Mon Sep 17 00:00:00 2001 From: Adeem Jassani Date: Fri, 22 May 2026 17:59:46 -0400 Subject: [PATCH 5/8] feat: vector-backed g_eventMap for HIP event handle reuse HIP event handles are routinely reused: applications call hipEventRecord on the same hipEvent_t many times across a trace. The previous single-entry map (hipEvent_t -> {stream, corrId}) collapsed all those records onto the most recent one, which misattributes the producer of a hipStreamWaitEvent whose own callback fires after a subsequent record on the same event. Switch g_eventMap to vector, keep the per-handle vector sorted by correlationId on insert via std::lower_bound, and resolve hipStreamWaitEvent by std::upper_bound + prev to pick the most recent record whose correlationId strictly precedes the wait's. Clear the map in RocmActivityProfiler::onResetTraceData so a session boundary cannot leak stale entries into the next trace. Mirrors the CUPTI design in CuptiActivityProfiler.cpp:waitEventMap() + updateWaitEventMap + getWaitEventInfo for CUPTI parity. Co-authored-by: Cursor --- libkineto/src/RocmActivityProfiler.cpp | 6 +++ libkineto/src/RocprofLogger.cpp | 54 +++++++++++++++++++++++--- libkineto/src/RocprofLogger.h | 6 +++ 3 files changed, 60 insertions(+), 6 deletions(-) diff --git a/libkineto/src/RocmActivityProfiler.cpp b/libkineto/src/RocmActivityProfiler.cpp index 9999013ac..ddbd03eb0 100644 --- a/libkineto/src/RocmActivityProfiler.cpp +++ b/libkineto/src/RocmActivityProfiler.cpp @@ -162,6 +162,12 @@ void RocmActivityProfiler::popCorrelationIdImpl(CorrelationFlowType type) { void RocmActivityProfiler::onResetTraceData() { roc_.teardownContext(); +#ifndef ROCTRACER_FALLBACK + // Drop any hipEvent_t -> {stream, corrId} entries left over from the prior + // profiling session so they cannot be returned as the producer of a wait + // recorded in the next session. Mirrors CuptiActivityProfiler::onResetTraceData. + RocprofLogger::clearEventMap(); +#endif } void RocmActivityProfiler::onFinalizeTrace( diff --git a/libkineto/src/RocprofLogger.cpp b/libkineto/src/RocprofLogger.cpp index 49ad52c5e..30813ad3c 100644 --- a/libkineto/src/RocprofLogger.cpp +++ b/libkineto/src/RocprofLogger.cpp @@ -17,9 +17,11 @@ #include #include +#include #include #include #include +#include #include "ApproximateClock.h" #include "Demangle.h" @@ -325,13 +327,25 @@ auto extract_sync_args = return 0; }; -// Maps hipEvent_t -> {hipStream_t, correlation_id} from hipEventRecord calls +// Maps hipEvent_t -> sorted vector of {hipStream_t, correlation_id} for every +// hipEventRecord observed on that event handle. +// +// HIP event handles are reused: applications routinely call hipEventRecord on +// the same hipEvent_t many times across the trace. Storing only the most +// recent record (single-entry map) gives wrong attribution when an earlier +// hipStreamWaitEvent fires its callback after a later hipEventRecord (cudaEvent +// callbacks are not guaranteed in-order across streams). +// +// The vector is kept sorted by correlationId so that a hipStreamWaitEvent +// callback can binary-search for the most recent record whose correlationId is +// strictly less than its own (i.e., the producer record it actually waits on). +// This mirrors CUPTI's `waitEventMap` design in CuptiActivityProfiler.cpp. struct EventMapEntry { hipStream_t stream{nullptr}; uint64_t corrId{0}; }; std::mutex g_eventMapMutex; -std::unordered_map g_eventMap; +std::unordered_map> g_eventMap; class RocprofApiIdList : public ApiIdList { public: @@ -598,6 +612,11 @@ void RocprofLogger::popCorrelationID(CorrelationDomain type) { } } +void RocprofLogger::clearEventMap() { + std::lock_guard lock(g_eventMapMutex); + g_eventMap.clear(); +} + void RocprofLogger::clearLogs() { // CuptiActivityProfiler clears this before the output Loggers use the data // for (auto &row : rows_) @@ -763,8 +782,16 @@ void RocprofLogger::api_callback( { std::lock_guard lock(g_eventMapMutex); - g_eventMap[static_cast(args.event)] = - {args.stream, record.correlation_id.internal}; + auto& vec = g_eventMap[static_cast(args.event)]; + EventMapEntry entry{args.stream, record.correlation_id.internal}; + auto pos = std::lower_bound( + vec.begin(), + vec.end(), + entry.corrId, + [](const EventMapEntry& a, uint64_t val) { + return a.corrId < val; + }); + vec.insert(pos, entry); } rocprofEventRecordRow* row = new rocprofEventRecordRow( @@ -797,8 +824,23 @@ void RocprofLogger::api_callback( std::lock_guard lock(g_eventMapMutex); auto it = g_eventMap.find(static_cast(args.event)); if (it != g_eventMap.end()) { - srcStream = it->second.stream; - srcCorrId = it->second.corrId; + const auto& vec = it->second; + // Find the most recent record whose correlationId is strictly + // less than the wait's own correlationId. The vector is sorted by + // correlationId, so upper_bound + prev gives that record. + uint64_t queryCorrId = record.correlation_id.internal; + auto pos = std::upper_bound( + vec.begin(), + vec.end(), + queryCorrId, + [](uint64_t val, const EventMapEntry& a) { + return val < a.corrId; + }); + if (pos != vec.begin()) { + auto prev = std::prev(pos); + srcStream = prev->stream; + srcCorrId = prev->corrId; + } } break; } diff --git a/libkineto/src/RocprofLogger.h b/libkineto/src/RocprofLogger.h index a13d8d0c9..b5dc86846 100644 --- a/libkineto/src/RocprofLogger.h +++ b/libkineto/src/RocprofLogger.h @@ -37,6 +37,12 @@ class RocprofLogger { static void pushCorrelationID(uint64_t id, RocLogger::CorrelationDomain type); static void popCorrelationID(RocLogger::CorrelationDomain type); + // Clears the global hipEvent_t -> {stream, correlationId} map populated by + // hipEventRecord callbacks. Must be called between profiling sessions + // (typically from RocmActivityProfiler::onResetTraceData) to prevent stale + // entries from a previous trace polluting the next one. + static void clearEventMap(); + static void ensureRegistered(); void startLogging(); void stopLogging(); From f4d594d266882fb5fa689cde636d085354767b17 Mon Sep 17 00:00:00 2001 From: Adeem Jassani Date: Fri, 22 May 2026 18:05:39 -0400 Subject: [PATCH 6/8] feat: extend dep resolution to hipEventSynchronize + emit wait_on_hip_event_id Two CUPTI-parity gaps in the rocprofiler-sdk sync metadata: 1. Only `hipStreamWaitEvent` looked up `g_eventMap` to attribute the producer record. `hipEventSynchronize`, which on the CUPTI side falls into the same `isEventSync()` bucket (STREAM_WAIT_EVENT and EVENT_SYNCHRONIZE both consult `waitEventMap`), got no producer attribution at all. Refactor the lookup out of the switch so both sync types share it. 2. The emitted JSON lacked a `wait_on_hip_event_id` field. CUPTI's `eventSyncInfo` always emits `wait_on_cuda_event_id` (the event handle itself) regardless of whether a producer record was found, so tooling can correlate waits even when the corresponding record fell outside the trace window. Mirror that: always emit `wait_on_hip_event_id` for the two event-sync types, and add the producer pair (`wait_on_stream`, `wait_on_hip_event_record_corr_id`) only when `g_eventMap` resolved one. Field-name map vs CUPTI is now: wait_on_stream <=> wait_on_stream wait_on_hip_event_record_corr_id <=> wait_on_cuda_event_record_corr_id wait_on_hip_event_id <=> wait_on_cuda_event_id Co-authored-by: Cursor --- libkineto/src/RocprofActivity_inl.h | 25 ++++++++++++-- libkineto/src/RocprofLogger.cpp | 52 ++++++++++++++++------------- 2 files changed, 51 insertions(+), 26 deletions(-) diff --git a/libkineto/src/RocprofActivity_inl.h b/libkineto/src/RocprofActivity_inl.h index dd8e4fb15..d46f45d45 100644 --- a/libkineto/src/RocprofActivity_inl.h +++ b/libkineto/src/RocprofActivity_inl.h @@ -282,13 +282,32 @@ inline const std::string RuntimeActivity::metadataJson() const { "hip_event": "{}")JSON", fmt::ptr(raw().event)); } - if (raw().syncType == ROCPROF_SYNC_STREAM_WAIT_EVENT && raw().srcCorrId) { + // Inter-stream dependency metadata: emitted for sync types that wait on a + // specific hipEvent_t (stream wait event, event synchronize) whenever the + // event was resolved against a prior hipEventRecord in g_eventMap. Field + // names mirror CUPTI's `wait_on_*` keys for CuptiActivityProfiler parity: + // + // wait_on_stream <=> CUPTI wait_on_stream + // wait_on_hip_event_record_corr_id <=> CUPTI wait_on_cuda_event_record_corr_id + // wait_on_hip_event_id <=> CUPTI wait_on_cuda_event_id + // + // The last field reports the hipEvent_t handle the wait was issued against, + // independent of whether a producer record was found. + if ((raw().syncType == ROCPROF_SYNC_STREAM_WAIT_EVENT || + raw().syncType == ROCPROF_SYNC_EVENT_SYNCHRONIZE) && + raw().event) { meta += fmt::format( R"JSON(, + "wait_on_hip_event_id": "{}")JSON", + fmt::ptr(raw().event)); + if (raw().srcCorrId) { + meta += fmt::format( + R"JSON(, "wait_on_stream": "{}", "wait_on_hip_event_record_corr_id": {})JSON", - fmt::ptr(raw().srcStream), - raw().srcCorrId); + fmt::ptr(raw().srcStream), + raw().srcCorrId); + } } return meta; } diff --git a/libkineto/src/RocprofLogger.cpp b/libkineto/src/RocprofLogger.cpp index 30813ad3c..ca95e183a 100644 --- a/libkineto/src/RocprofLogger.cpp +++ b/libkineto/src/RocprofLogger.cpp @@ -819,31 +819,8 @@ void RocprofLogger::api_callback( switch (record.operation) { case ROCPROFILER_HIP_RUNTIME_API_ID_hipStreamWaitEvent: case ROCPROFILER_HIP_RUNTIME_API_ID_hipStreamWaitEvent_spt: - { syncType = ROCPROF_SYNC_STREAM_WAIT_EVENT; - std::lock_guard lock(g_eventMapMutex); - auto it = g_eventMap.find(static_cast(args.event)); - if (it != g_eventMap.end()) { - const auto& vec = it->second; - // Find the most recent record whose correlationId is strictly - // less than the wait's own correlationId. The vector is sorted by - // correlationId, so upper_bound + prev gives that record. - uint64_t queryCorrId = record.correlation_id.internal; - auto pos = std::upper_bound( - vec.begin(), - vec.end(), - queryCorrId, - [](uint64_t val, const EventMapEntry& a) { - return val < a.corrId; - }); - if (pos != vec.begin()) { - auto prev = std::prev(pos); - srcStream = prev->stream; - srcCorrId = prev->corrId; - } - } break; - } case ROCPROFILER_HIP_RUNTIME_API_ID_hipEventSynchronize: syncType = ROCPROF_SYNC_EVENT_SYNCHRONIZE; break; @@ -857,6 +834,35 @@ void RocprofLogger::api_callback( break; } + // For sync types that wait on a specific hipEvent_t (stream wait event + // and event synchronize), look up the most recent hipEventRecord on + // that event whose correlationId strictly precedes this sync's, so we + // can emit the producer stream + corr_id in the trace metadata. + // Matches CUPTI's isEventSync() handling for STREAM_WAIT_EVENT and + // EVENT_SYNCHRONIZE. + if ((syncType == ROCPROF_SYNC_STREAM_WAIT_EVENT || + syncType == ROCPROF_SYNC_EVENT_SYNCHRONIZE) && + args.event != nullptr) { + std::lock_guard lock(g_eventMapMutex); + auto it = g_eventMap.find(static_cast(args.event)); + if (it != g_eventMap.end()) { + const auto& vec = it->second; + uint64_t queryCorrId = record.correlation_id.internal; + auto pos = std::upper_bound( + vec.begin(), + vec.end(), + queryCorrId, + [](uint64_t val, const EventMapEntry& a) { + return val < a.corrId; + }); + if (pos != vec.begin()) { + auto prev = std::prev(pos); + srcStream = prev->stream; + srcCorrId = prev->corrId; + } + } + } + rocprofSyncRow* row = new rocprofSyncRow( record.correlation_id.internal, record.kind, From 925f99bb385ce78135dcac3f5835ffb0a395f992 Mon Sep 17 00:00:00 2001 From: Adeem Jassani Date: Fri, 22 May 2026 18:10:28 -0400 Subject: [PATCH 7/8] test: cover hipEvent_t reuse + reset + event sync + unresolved waits Adds four new tests that mirror CUPTI's wait-event coverage in CuptiActivityProfilerTest.cpp: StreamWaitEventFutureCorrelation Two hipEventRecord callbacks on the same hipEvent_t (corr 100 then 200) and a hipStreamWaitEvent at corr 101 in between. Without the vector-backed g_eventMap, the wait would be attributed to the more recent (corr 200) record; with it, the wait correctly resolves to corr 100. Exercises lower_bound insert + upper_bound lookup. EventMapClearedOnReset Records an event in session 1, drives RocmActivityProfiler::reset(), then issues a wait in session 2 with no prior record. The wait must NOT pick up session 1's stale entry. EventSynchronizeResolvesProducer Records an event and then calls hipEventSynchronize. Verifies that the sync gets producer attribution (previously only hipStreamWaitEvent did) and that wait_on_hip_event_id is emitted. UnresolvedWaitStillEmitsEventId Wait fires on an event whose record predated the trace window. Verifies wait_on_hip_event_id is still emitted (tooling can match waits against records observed elsewhere) while wait_on_hip_event_record_corr_id / wait_on_stream are omitted. To drive the production lookup path from tests, the inline g_eventMap operations in the api_callback were extracted into three static helpers on RocprofLogger: recordEvent(event, stream, corrId) resolveWait(event, queryCorrId, &outStream, &outCorrId) clearEventMap() MockRocLogger now calls recordEvent() from addEventRecordActivity() and exposes addSyncActivityResolvingFromMap() so test traces exercise the exact same lookup the real callback does. Co-authored-by: Cursor --- libkineto/src/RocprofLogger.cpp | 87 ++++-- libkineto/src/RocprofLogger.h | 18 ++ libkineto/test/RocmActivityProfilerTest.cpp | 290 +++++++++++++++++++- 3 files changed, 364 insertions(+), 31 deletions(-) diff --git a/libkineto/src/RocprofLogger.cpp b/libkineto/src/RocprofLogger.cpp index ca95e183a..f54ec01b5 100644 --- a/libkineto/src/RocprofLogger.cpp +++ b/libkineto/src/RocprofLogger.cpp @@ -612,6 +612,50 @@ void RocprofLogger::popCorrelationID(CorrelationDomain type) { } } +void RocprofLogger::recordEvent( + void* event, + void* stream, + uint64_t corrId) { + std::lock_guard lock(g_eventMapMutex); + auto& vec = g_eventMap[event]; + EventMapEntry entry{static_cast(stream), corrId}; + auto pos = std::lower_bound( + vec.begin(), + vec.end(), + entry.corrId, + [](const EventMapEntry& a, uint64_t val) { return a.corrId < val; }); + vec.insert(pos, entry); +} + +bool RocprofLogger::resolveWait( + void* event, + uint64_t queryCorrId, + void** outStream, + uint64_t* outCorrId) { + std::lock_guard lock(g_eventMapMutex); + auto it = g_eventMap.find(event); + if (it == g_eventMap.end()) { + return false; + } + const auto& vec = it->second; + auto pos = std::upper_bound( + vec.begin(), + vec.end(), + queryCorrId, + [](uint64_t val, const EventMapEntry& a) { return val < a.corrId; }); + if (pos == vec.begin()) { + return false; + } + auto prev = std::prev(pos); + if (outStream) { + *outStream = static_cast(prev->stream); + } + if (outCorrId) { + *outCorrId = prev->corrId; + } + return true; +} + void RocprofLogger::clearEventMap() { std::lock_guard lock(g_eventMapMutex); g_eventMap.clear(); @@ -780,19 +824,10 @@ void RocprofLogger::api_callback( rocprofiler_iterate_callback_tracing_kind_operation_args( record, extract_event_record_args, 1, &args); - { - std::lock_guard lock(g_eventMapMutex); - auto& vec = g_eventMap[static_cast(args.event)]; - EventMapEntry entry{args.stream, record.correlation_id.internal}; - auto pos = std::lower_bound( - vec.begin(), - vec.end(), - entry.corrId, - [](const EventMapEntry& a, uint64_t val) { - return a.corrId < val; - }); - vec.insert(pos, entry); - } + recordEvent( + static_cast(args.event), + static_cast(args.stream), + record.correlation_id.internal); rocprofEventRecordRow* row = new rocprofEventRecordRow( record.correlation_id.internal, @@ -843,23 +878,15 @@ void RocprofLogger::api_callback( if ((syncType == ROCPROF_SYNC_STREAM_WAIT_EVENT || syncType == ROCPROF_SYNC_EVENT_SYNCHRONIZE) && args.event != nullptr) { - std::lock_guard lock(g_eventMapMutex); - auto it = g_eventMap.find(static_cast(args.event)); - if (it != g_eventMap.end()) { - const auto& vec = it->second; - uint64_t queryCorrId = record.correlation_id.internal; - auto pos = std::upper_bound( - vec.begin(), - vec.end(), - queryCorrId, - [](uint64_t val, const EventMapEntry& a) { - return val < a.corrId; - }); - if (pos != vec.begin()) { - auto prev = std::prev(pos); - srcStream = prev->stream; - srcCorrId = prev->corrId; - } + void* resolvedStream = nullptr; + uint64_t resolvedCorrId = 0; + if (resolveWait( + static_cast(args.event), + record.correlation_id.internal, + &resolvedStream, + &resolvedCorrId)) { + srcStream = static_cast(resolvedStream); + srcCorrId = resolvedCorrId; } } diff --git a/libkineto/src/RocprofLogger.h b/libkineto/src/RocprofLogger.h index b5dc86846..8fe331d54 100644 --- a/libkineto/src/RocprofLogger.h +++ b/libkineto/src/RocprofLogger.h @@ -37,6 +37,24 @@ class RocprofLogger { static void pushCorrelationID(uint64_t id, RocLogger::CorrelationDomain type); static void popCorrelationID(RocLogger::CorrelationDomain type); + // Insert a hipEventRecord observation into the global hipEvent_t -> + // sorted vector<{stream, correlationId}> map. Called from the + // hipEventRecord callback to remember the producer for a later + // hipStreamWaitEvent / hipEventSynchronize lookup. Test fixtures may + // also call this directly to seed the map without going through the + // real ROCm tracing path. + static void recordEvent(void* event, void* stream, uint64_t corrId); + + // Look up the most recent hipEventRecord observation for `event` whose + // correlationId is strictly less than `queryCorrId`. On a successful + // lookup, sets *outStream / *outCorrId and returns true. Returns false + // when the event has no record (or no record older than the query). + static bool resolveWait( + void* event, + uint64_t queryCorrId, + void** outStream, + uint64_t* outCorrId); + // Clears the global hipEvent_t -> {stream, correlationId} map populated by // hipEventRecord callbacks. Must be called between profiling sessions // (typically from RocmActivityProfiler::onResetTraceData) to prevent stale diff --git a/libkineto/test/RocmActivityProfilerTest.cpp b/libkineto/test/RocmActivityProfilerTest.cpp index 85d735ade..02403cd04 100644 --- a/libkineto/test/RocmActivityProfilerTest.cpp +++ b/libkineto/test/RocmActivityProfilerTest.cpp @@ -311,6 +311,10 @@ struct MockRocLogger { } #ifndef ROCTRACER_FALLBACK + // Adds a hipEventRecord activity AND populates RocprofLogger's global + // hipEvent_t -> {stream, corrId} map (just like the real api_callback + // would). New tests should prefer addSyncActivityResolvingFromMap() so + // the JSON output goes through the production lookup path. void addEventRecordActivity( int64_t start_ns, int64_t end_ns, @@ -328,8 +332,14 @@ struct MockRocLogger { event, stream); activities_.push_back(row); + RocprofLogger::recordEvent( + static_cast(event), + static_cast(stream), + static_cast(correlation)); } + // Pre-resolved variant: pass producer (srcStream, srcCorrId) directly. + // Used by tests that already know what the lookup should return. void addSyncActivity( int64_t start_ns, int64_t end_ns, @@ -339,10 +349,14 @@ struct MockRocLogger { hipEvent_t event, hipStream_t srcStream, uint64_t srcCorrId) { + uint32_t apiId = + (syncType == ROCPROF_SYNC_EVENT_SYNCHRONIZE) + ? ROCPROFILER_HIP_RUNTIME_API_ID_hipEventSynchronize + : ROCPROFILER_HIP_RUNTIME_API_ID_hipStreamWaitEvent; rocprofSyncRow* row = new rocprofSyncRow( correlation, RUNTIME_DOMAIN, - ROCPROFILER_HIP_RUNTIME_API_ID_hipStreamWaitEvent, + apiId, processId(), systemThreadId(), start_ns, @@ -354,6 +368,37 @@ struct MockRocLogger { srcCorrId); activities_.push_back(row); } + + // Lookup-driven variant: queries RocprofLogger's global event map for + // the producer record matching `event` whose correlationId is < `correlation`, + // matching the production api_callback path. Tests B1 (vector + upper_bound) + // and B3 (event sync resolution) end-to-end through the JSON output. + void addSyncActivityResolvingFromMap( + int64_t start_ns, + int64_t end_ns, + int64_t correlation, + rocprofSyncType syncType, + hipStream_t stream, + hipEvent_t event) { + hipStream_t srcStream = nullptr; + uint64_t srcCorrId = 0; + if (syncType == ROCPROF_SYNC_STREAM_WAIT_EVENT || + syncType == ROCPROF_SYNC_EVENT_SYNCHRONIZE) { + void* resolvedStream = nullptr; + uint64_t resolvedCorrId = 0; + if (RocprofLogger::resolveWait( + static_cast(event), + static_cast(correlation), + &resolvedStream, + &resolvedCorrId)) { + srcStream = static_cast(resolvedStream); + srcCorrId = resolvedCorrId; + } + } + addSyncActivity( + start_ns, end_ns, correlation, syncType, stream, event, srcStream, + srcCorrId); + } #endif ~MockRocLogger() { @@ -1178,4 +1223,247 @@ TEST_F(RocmActivityProfilerTest, InterStreamDependencyTest) { unlink(filename); } + +// Drives the production lookup helpers (RocprofLogger::recordEvent / +// resolveWait / clearEventMap) so we exercise the same path used by the +// rocprofiler-sdk api_callback. Verifies B1's vector-backed g_eventMap and +// the upper_bound semantics on event-handle reuse. +TEST_F(RocmActivityProfilerTest, StreamWaitEventFutureCorrelation) { + // Out-of-order delivery: two hipEventRecord callbacks land on the same + // event handle (corr=100 then corr=200), and a hipStreamWaitEvent with + // corr=101 lands AFTER both records. The wait should attribute its + // producer to corr=100 (most recent record < 101), not corr=200. + RocprofLogger::clearEventMap(); + + RocmActivityProfiler profiler(rocActivities_, /*cpu only*/ false); + int64_t start_time_ns = + libkineto::timeSinceEpoch(std::chrono::system_clock::now()); + int64_t duration_ns = 500; + auto start_time = time_point(nanoseconds(start_time_ns)); + profiler.configure(*cfg_, start_time); + profiler.startTrace(start_time); + profiler.stopTrace(start_time + nanoseconds(duration_ns)); + profiler.recordThreadInfo(); + + auto cpuOps = std::make_unique( + start_time_ns, start_time_ns + duration_ns); + cpuOps->addOp("op1", start_time_ns + 10, start_time_ns + 30, 1); + profiler.transferCpuTrace(std::move(cpuOps)); + + auto gpuOps = std::make_unique(); + hipEvent_t fakeEvent = reinterpret_cast(0xA); + hipStream_t producer1 = reinterpret_cast(0x1); + hipStream_t producer2 = reinterpret_cast(0x2); + hipStream_t consumer = reinterpret_cast(0x3); + + // Two records on the same event handle, corr=100 and corr=200. + gpuOps->addEventRecordActivity( + start_time_ns + 20, start_time_ns + 25, 100, fakeEvent, producer1); + gpuOps->addEventRecordActivity( + start_time_ns + 60, start_time_ns + 65, 200, fakeEvent, producer2); + // Wait at corr=101 must resolve to corr=100, not corr=200. + gpuOps->addSyncActivityResolvingFromMap( + start_time_ns + 40, start_time_ns + 45, 101, + ROCPROF_SYNC_STREAM_WAIT_EVENT, consumer, fakeEvent); + rocActivities_.activityLogger = std::move(gpuOps); + + auto logger = std::make_unique(*cfg_); + profiler.processTrace(*logger); + profiler.reset(); + + ActivityTrace trace(std::move(logger), loggerFactory); + + char filename[] = "/tmp/libkineto_future_corrXXXXXX.json"; + { int tmp_fd = mkstemps(filename, 5); + if (tmp_fd >= 0) close(tmp_fd); } + trace.save(filename); + + std::ifstream f(filename); + nlohmann::json j = nlohmann::json::parse(f); + bool foundWait = false; + for (auto& ev : j["traceEvents"]) { + if (ev.value("name", "") == "hipStreamWaitEvent" && + ev.contains("args") && + ev["args"].contains("wait_on_hip_event_record_corr_id")) { + foundWait = true; + EXPECT_EQ(ev["args"]["wait_on_hip_event_record_corr_id"], 100) + << "Wait should reference the record before it, not the future one"; + } + } + EXPECT_TRUE(foundWait); + unlink(filename); + + RocprofLogger::clearEventMap(); +} + +// Verifies B1's clear-on-reset behavior: records from a prior profiling +// session must not leak into the next session's wait resolution. +TEST_F(RocmActivityProfilerTest, EventMapClearedOnReset) { + RocprofLogger::clearEventMap(); + + hipEvent_t fakeEvent = reinterpret_cast(0xB); + hipStream_t producer = reinterpret_cast(0x1); + + // Session 1: seed g_eventMap with a record, then call onResetTraceData + // (via profiler.reset()) which should clear it. + { + RocmActivityProfiler profiler(rocActivities_, /*cpu only*/ false); + int64_t start_time_ns = + libkineto::timeSinceEpoch(std::chrono::system_clock::now()); + auto start_time = time_point(nanoseconds(start_time_ns)); + profiler.configure(*cfg_, start_time); + profiler.startTrace(start_time); + profiler.stopTrace(start_time + nanoseconds(500)); + profiler.recordThreadInfo(); + + auto cpuOps = std::make_unique( + start_time_ns, start_time_ns + 500); + cpuOps->addOp("op1", start_time_ns + 10, start_time_ns + 30, 1); + profiler.transferCpuTrace(std::move(cpuOps)); + + auto gpuOps = std::make_unique(); + gpuOps->addEventRecordActivity( + start_time_ns + 20, start_time_ns + 25, 50, fakeEvent, producer); + rocActivities_.activityLogger = std::move(gpuOps); + + auto logger = std::make_unique(*cfg_); + profiler.processTrace(*logger); + profiler.reset(); + } + + // Session 2: wait on the same event handle. With clearEventMap() called + // on reset, the wait must NOT resolve to session 1's stale record. + void* outStream = nullptr; + uint64_t outCorrId = 0; + EXPECT_FALSE(RocprofLogger::resolveWait( + static_cast(fakeEvent), 999, &outStream, &outCorrId)) + << "Stale record from a prior session should have been cleared"; +} + +// Verifies B3's CUPTI parity: hipEventSynchronize gets producer +// attribution just like hipStreamWaitEvent, and the always-emitted +// wait_on_hip_event_id field is present. +TEST_F(RocmActivityProfilerTest, EventSynchronizeResolvesProducer) { + RocprofLogger::clearEventMap(); + + RocmActivityProfiler profiler(rocActivities_, /*cpu only*/ false); + int64_t start_time_ns = + libkineto::timeSinceEpoch(std::chrono::system_clock::now()); + int64_t duration_ns = 300; + auto start_time = time_point(nanoseconds(start_time_ns)); + profiler.configure(*cfg_, start_time); + profiler.startTrace(start_time); + profiler.stopTrace(start_time + nanoseconds(duration_ns)); + profiler.recordThreadInfo(); + + auto cpuOps = std::make_unique( + start_time_ns, start_time_ns + duration_ns); + cpuOps->addOp("op1", start_time_ns + 10, start_time_ns + 30, 1); + profiler.transferCpuTrace(std::move(cpuOps)); + + auto gpuOps = std::make_unique(); + hipEvent_t fakeEvent = reinterpret_cast(0xC); + hipStream_t producer = reinterpret_cast(0x1); + // hipEventSynchronize is consumer-stream-less; pass nullptr for the + // consumer stream as the real API does. + hipStream_t consumer = nullptr; + + gpuOps->addEventRecordActivity( + start_time_ns + 20, start_time_ns + 25, 30, fakeEvent, producer); + gpuOps->addSyncActivityResolvingFromMap( + start_time_ns + 50, start_time_ns + 60, 40, + ROCPROF_SYNC_EVENT_SYNCHRONIZE, consumer, fakeEvent); + rocActivities_.activityLogger = std::move(gpuOps); + + auto logger = std::make_unique(*cfg_); + profiler.processTrace(*logger); + profiler.reset(); + + ActivityTrace trace(std::move(logger), loggerFactory); + + char filename[] = "/tmp/libkineto_event_syncXXXXXX.json"; + { int tmp_fd = mkstemps(filename, 5); + if (tmp_fd >= 0) close(tmp_fd); } + trace.save(filename); + + std::ifstream f(filename); + nlohmann::json j = nlohmann::json::parse(f); + bool foundEventSync = false; + for (auto& ev : j["traceEvents"]) { + if (ev.value("name", "") == "hipEventSynchronize" && + ev.contains("args")) { + foundEventSync = true; + EXPECT_EQ(ev["args"]["sync_type"], "event_synchronize"); + // wait_on_hip_event_id is always emitted for event-sync types + EXPECT_TRUE(ev["args"].contains("wait_on_hip_event_id")); + // Producer attribution resolved from g_eventMap + EXPECT_TRUE(ev["args"].contains("wait_on_hip_event_record_corr_id")); + EXPECT_EQ(ev["args"]["wait_on_hip_event_record_corr_id"], 30); + EXPECT_TRUE(ev["args"].contains("wait_on_stream")); + } + } + EXPECT_TRUE(foundEventSync) << "hipEventSynchronize activity not found"; + unlink(filename); + + RocprofLogger::clearEventMap(); +} + +// Verifies wait_on_hip_event_id is emitted even when the producer record is +// absent (e.g., recorded before the trace window). +TEST_F(RocmActivityProfilerTest, UnresolvedWaitStillEmitsEventId) { + RocprofLogger::clearEventMap(); + + RocmActivityProfiler profiler(rocActivities_, /*cpu only*/ false); + int64_t start_time_ns = + libkineto::timeSinceEpoch(std::chrono::system_clock::now()); + int64_t duration_ns = 300; + auto start_time = time_point(nanoseconds(start_time_ns)); + profiler.configure(*cfg_, start_time); + profiler.startTrace(start_time); + profiler.stopTrace(start_time + nanoseconds(duration_ns)); + profiler.recordThreadInfo(); + + auto cpuOps = std::make_unique( + start_time_ns, start_time_ns + duration_ns); + cpuOps->addOp("op1", start_time_ns + 10, start_time_ns + 30, 1); + profiler.transferCpuTrace(std::move(cpuOps)); + + // No addEventRecordActivity: the wait fires on an event that was never + // observed in this trace window. + auto gpuOps = std::make_unique(); + hipEvent_t fakeEvent = reinterpret_cast(0xD); + hipStream_t consumer = reinterpret_cast(0x2); + gpuOps->addSyncActivityResolvingFromMap( + start_time_ns + 50, start_time_ns + 60, 70, + ROCPROF_SYNC_STREAM_WAIT_EVENT, consumer, fakeEvent); + rocActivities_.activityLogger = std::move(gpuOps); + + auto logger = std::make_unique(*cfg_); + profiler.processTrace(*logger); + profiler.reset(); + + ActivityTrace trace(std::move(logger), loggerFactory); + + char filename[] = "/tmp/libkineto_unresolved_waitXXXXXX.json"; + { int tmp_fd = mkstemps(filename, 5); + if (tmp_fd >= 0) close(tmp_fd); } + trace.save(filename); + + std::ifstream f(filename); + nlohmann::json j = nlohmann::json::parse(f); + bool foundUnresolvedWait = false; + for (auto& ev : j["traceEvents"]) { + if (ev.value("name", "") == "hipStreamWaitEvent" && + ev.contains("args")) { + foundUnresolvedWait = true; + // Event id must still be emitted for tooling correlation. + EXPECT_TRUE(ev["args"].contains("wait_on_hip_event_id")); + // Producer attribution must be absent. + EXPECT_FALSE(ev["args"].contains("wait_on_hip_event_record_corr_id")); + EXPECT_FALSE(ev["args"].contains("wait_on_stream")); + } + } + EXPECT_TRUE(foundUnresolvedWait); + unlink(filename); +} #endif From e097ad6e24cca7b1b243f39c4d56ac872181a1d1 Mon Sep 17 00:00:00 2001 From: Adeem Jassani Date: Fri, 22 May 2026 18:31:16 -0400 Subject: [PATCH 8/8] build: suppress -Wpedantic on ROCm builds against system rocprofiler-sdk The rocprofiler-sdk and HSA headers shipped with ROCm 7.x use GCC extensions (anonymous structs, flexible array members) that violate strict ISO C++. When the consumer build propagates `-Wpedantic` to kineto - PyTorch's cmake/Dependencies.cmake does this for everything under caffe2_* - compilation of RocprofLogger.cpp, RocprofActivityApi.cpp and init.cpp fails with errors like: /opt/rocm-7.2.3/include/rocprofiler-sdk/fwd.h:757:9: error: ISO C++ prohibits anonymous structs [-Werror=pedantic] These headers are owned by ROCm, not by us, so the fix lives on the consumer side. Add `-Wno-pedantic` to KINETO_COMPILE_OPTIONS for the rocm backend only; this turns off pedantic warnings for the kineto translation units that include the system headers, leaving the rest of the build untouched. Limiting the suppression to GCC/Clang avoids touching MSVC builds. Co-authored-by: Cursor --- libkineto/CMakeLists.txt | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/libkineto/CMakeLists.txt b/libkineto/CMakeLists.txt index f17f8b072..519b7f36e 100644 --- a/libkineto/CMakeLists.txt +++ b/libkineto/CMakeLists.txt @@ -205,6 +205,14 @@ elseif(KINETO_BACKEND STREQUAL "rocm") endif() target_compile_definitions(kineto_base PRIVATE "__HIP_PLATFORM_HCC__") target_compile_definitions(kineto_base PRIVATE "__HIP_PLATFORM_AMD__") + # The rocprofiler-sdk and HSA system headers (anonymous structs, flexible + # array members, etc.) trip -Werror=pedantic when callers (e.g. PyTorch's + # cmake/Dependencies.cmake) propagate -Wpedantic to us. Silence pedantic for + # the kineto sources only; the headers themselves come from /opt/rocm and + # we can't change them. + if(CMAKE_CXX_COMPILER_ID MATCHES "GNU|Clang") + list(APPEND KINETO_COMPILE_OPTIONS "-Wno-pedantic") + endif() elseif(KINETO_BACKEND STREQUAL "xpu") list(APPEND KINETO_COMPILE_OPTIONS ${XPUPTI_BUILD_FLAG}) if(KINETO_BUILD_TESTS)