Line data Source code
1 : /**
2 : * Copyright (c) 2025 Huawei Technologies Co., Ltd.
3 : * This program is free software, you can redistribute it and/or modify it under the terms and conditions of
4 : * CANN Open Software License Agreement Version 2.0 (the "License").
5 : * Please refer to the License for details. You may not use this file except in compliance with the License.
6 : * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,
7 : * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
8 : * See LICENSE in the root of the software repository for the full text of the License.
9 : */
10 : #include "hcclCommDfx.h"
11 : #include "ccu_rep_context_v1.h"
12 : #include "task_info.h"
13 :
14 : namespace hccl {
15 :
16 : std::shared_mutex HcclCommDfx::baseLock_;
17 : std::mutex HcclCommDfx::taskIdMutex_;
18 : std::unordered_map<std::string, std::unordered_map<u64, u32>> HcclCommDfx::channelRemoteRankId_;
19 : std::unordered_map<u32, u32> HcclCommDfx::streamIdToTaskId_;
20 169 : HcclCommDfx::HcclCommDfx() {}
21 :
22 169 : HcclCommDfx::~HcclCommDfx()
23 : {
24 169 : setAddTaskCallback_ = nullptr;
25 169 : setAddDpuTaskCallback_ = nullptr;
26 169 : }
27 :
28 169 : HcclResult HcclCommDfx::Init(u32 deviceId, const std::string& comTag, u32 myRankId)
29 : {
30 169 : if (initializedFlag_) {
31 0 : return HCCL_SUCCESS;
32 : }
33 169 : HCCL_INFO("[%s]deviceId[%u], comTag[%s], myRankId[%u]", __func__, deviceId, comTag.c_str(), myRankId);
34 169 : deviceId_ = deviceId;
35 169 : commTag_ = comTag;
36 169 : myRankId_ = myRankId;
37 : // 1. 如果mirrorTaskManager_为空，则创建新的MirrorTaskManager
38 169 : if (!mirrorTaskManager_) {
39 : mirrorTaskManager_
40 169 : = std::make_unique<Hccl::MirrorTaskManager>(deviceId_, &Hccl::GlobalMirrorTasks::Instance(), false);
41 : }
42 :
43 : // 2. 创建Profiling管理类
44 169 : EXCEPTION_CATCH(
45 : profiling_ = std::make_unique<HcclCommProfiling>(deviceId_, mirrorTaskManager_.get()), return HCCL_E_PTR);
46 169 : CHK_RET(profiling_->Init());
47 :
48 : // 3. 注册回调
49 338 : setAddTaskCallback_ = [this](u32 streamId, u32 taskId, const Hccl::TaskParam& taskParam, u64 handle) {
50 0 : return this->AddTaskInfoCallback(streamId, taskId, taskParam, handle);
51 169 : };
52 339 : setAddDpuTaskCallback_ = [this](const Hccl::TaskParam& taskParam, u64 handle) {
53 1 : return this->AddDpuTaskInfoCallback(taskParam, handle);
54 169 : };
55 169 : initializedFlag_ = true;
56 169 : return HCCL_SUCCESS; // 初始化成功返回成功码
57 : }
58 :
59 2 : HcclResult HcclCommDfx::GetOpModeFlags(bool& isOpBase, bool& isCached)
60 : {
61 2 : auto currDfxOpInfo = mirrorTaskManager_->GetCurrDfxOpInfo();
62 2 : CHK_SMART_PTR_NULL(currDfxOpInfo);
63 2 : isOpBase = currDfxOpInfo->op_.opMode == Hccl::OpMode::OPBASE || currDfxOpInfo->op_.opMode == Hccl::OpMode::ACLGRAPH;
64 : isCached
65 2 : = currDfxOpInfo->op_.opMode == Hccl::OpMode::OFFLOAD || currDfxOpInfo->op_.opMode == Hccl::OpMode::ACLGRAPH;
66 2 : HCCL_INFO("[%s] GetOpModeFlags: isOpBase %d, isCached %d", __func__, isOpBase, isCached);
67 2 : return HCCL_SUCCESS;
68 2 : }
69 :
70 : // 回调注册实现
71 1 : void HcclCommDfx::AddTaskInfoCallbackLog(
72 : const Hccl::TaskParam& taskParam, const std::unordered_map<u64, u32>& handleMap) const
73 : {
74 1 : if (LIKELY(HcclCheckLogLevel(HCCL_LOG_INFO) == 0)) {
75 0 : return;
76 : }
77 2 : for (size_t i = 0; i < taskParam.ccuDetailInfo->size(); ++i) {
78 1 : const Hccl::CcuProfilingInfo& profInfo = (*taskParam.ccuDetailInfo)[i];
79 2 : for (int idx = 0; idx < hcomm::CCU_MAX_CHANNEL_NUM; idx++) {
80 2 : if (profInfo.channelId[idx] == hcomm::INVALID_VALUE_CHANNELID) {
81 1 : break;
82 : }
83 1 : auto handleIt = handleMap.find(profInfo.channelHandle[idx]);
84 1 : if (handleIt == handleMap.end()) {
85 0 : continue;
86 : }
87 1 : HCCL_INFO(
88 : "[%s]idx[%d]: channelId[%u], remoteRankId[%u], channelHandle[0x%llx]", __func__, idx,
89 : profInfo.channelId[idx], handleIt->second, profInfo.channelHandle[idx]);
90 : }
91 : }
92 : }
93 :
94 5 : HcclResult HcclCommDfx::AddTaskInfoCallback(u32 streamId, u32 taskId, const Hccl::TaskParam& taskParam, u64 handle)
95 : {
96 5 : u32 remoteRankId = INVALID_UINT;
97 5 : if (handle != DFX_INVALID_U64) {
98 1 : CHK_RET(GetChannelRemoteRankId(commTag_, handle, remoteRankId));
99 : }
100 :
101 5 : Hccl::PrintTaskLog(streamId, taskId, taskParam, remoteRankId);
102 :
103 5 : if (taskParam.taskType == Hccl::TaskParamType::TASK_CCU && taskParam.ccuDetailInfo != nullptr) {
104 2 : std::shared_lock<std::shared_mutex> rwLock(baseLock_);
105 2 : auto commIt = channelRemoteRankId_.find(commTag_);
106 2 : if (commIt == channelRemoteRankId_.end()) {
107 1 : HCCL_ERROR("[%s] commTag:[%s] not found in CCU batch lookup", __func__, commTag_.c_str());
108 1 : return HCCL_E_PARA;
109 : }
110 1 : const auto& handleMap = commIt->second;
111 1 : AddTaskInfoCallbackLog(taskParam, handleMap);
112 2 : for (size_t i = 0; i < taskParam.ccuDetailInfo->size(); ++i) {
113 1 : Hccl::CcuProfilingInfo& profInfo = (*taskParam.ccuDetailInfo)[i];
114 2 : for (int idx = 0; idx < hcomm::CCU_MAX_CHANNEL_NUM; idx++) {
115 2 : if (profInfo.channelId[idx] == hcomm::INVALID_VALUE_CHANNELID) {
116 1 : break;
117 : }
118 1 : auto handleIt = handleMap.find(profInfo.channelHandle[idx]);
119 1 : if (handleIt == handleMap.end()) {
120 0 : HCCL_ERROR(
121 : "[%s] Failed to get remote rank for channelHandle[0x%llx]", __func__,
122 : profInfo.channelHandle[idx]);
123 0 : return HCCL_E_PARA;
124 : }
125 1 : profInfo.remoteRankId[idx] = handleIt->second;
126 : }
127 : }
128 2 : }
129 8 : HcclResult ret = mirrorTaskManager_->AddTaskInfo(
130 8 : streamId, taskId, remoteRankId, taskParam, mirrorTaskManager_->GetCurrDfxOpInfo(), taskParam.isMaster);
131 4 : CHK_RET(ret);
132 4 : return HCCL_SUCCESS;
133 : }
134 :
135 3 : HcclResult HcclCommDfx::AddDpuTaskInfoCallback(const Hccl::TaskParam& taskParam, u64 handle)
136 : {
137 3 : u32 streamId = dpuStreamId_;
138 3 : u32 taskId = GetTaskId(streamId);
139 3 : Hccl::TaskParam localTaskParam = taskParam;
140 3 : localTaskParam.aicpuTaskId = aicpuTaskId_;
141 3 : localTaskParam.npuDevId = deviceId_;
142 3 : HCCL_INFO(
143 : "[%s] streamId[%u], taskId[%u], aicpuTaskId[%llu], npuDevId[%u].", __func__, streamId, taskId,
144 : localTaskParam.aicpuTaskId, static_cast<u32>(localTaskParam.npuDevId));
145 6 : return AddTaskInfoCallback(streamId, taskId, localTaskParam, handle);
146 3 : }
147 :
148 0 : HcclResult HcclCommDfx::SetCurrDfxOpInfo(std::shared_ptr<Hccl::DfxOpInfo> dfxOpInfo)
149 : {
150 0 : profiling_->SetCurrDfxOpInfo(dfxOpInfo);
151 0 : return HCCL_SUCCESS;
152 : }
153 :
154 : // HcclCommDfx接口实现 - 修改为返回HcclResult类型
155 146 : HcclResult HcclCommDfx::ReportAllTasks(bool cachedReq)
156 : {
157 146 : EXCEPTION_CATCH(profiling_->ReportAllTasks(cachedReq), return HCCL_E_PTR);
158 146 : return HCCL_SUCCESS;
159 : }
160 :
161 0 : HcclResult HcclCommDfx::ReportOp(uint64_t beginTime, bool cachedReq, bool isOpBase)
162 : {
163 0 : EXCEPTION_CATCH(profiling_->ReportOp(beginTime, cachedReq, isOpBase), return HCCL_E_PTR);
164 0 : return HCCL_SUCCESS;
165 : }
166 :
167 : // 返回值Mc2要改
168 5 : void HcclCommDfx::ReportMc2CommInfo(const Mc2CommInfo& mc2CommInfo) { profiling_->ReportMc2CommInfo(mc2CommInfo); }
169 :
170 0 : HcclResult HcclCommDfx::UpdateProfStat()
171 : {
172 0 : profiling_->UpdateProfStat();
173 0 : return HCCL_SUCCESS;
174 : }
175 :
176 0 : Hccl::MirrorTaskManager* HcclCommDfx::GetMirrorTaskManager() const { return mirrorTaskManager_.get(); }
177 :
178 : // 将remoteRankId添加到channelRemoteRankId_表中
179 5 : void HcclCommDfx::AddChannelRemoteRankId(const std::string& commTag, u64 handle, u32 remoteRankId)
180 : {
181 5 : std::unique_lock<std::shared_mutex> rwLock(baseLock_);
182 5 : HCCL_INFO(
183 : "[HcclCommDfx][AddChannelRemoteRankId] commTag:[%s], handle:[%llu], remoteRankId:[%u]", commTag.c_str(), handle,
184 : remoteRankId);
185 5 : channelRemoteRankId_[commTag][handle] = remoteRankId;
186 5 : }
187 :
188 : // 在channelRemoteRankId_表中对remoteRankId进行查找（原有逻辑补充返回值）
189 4 : HcclResult HcclCommDfx::GetChannelRemoteRankId(const std::string& commTag, u64 handle, u32& remoteRankId)
190 : {
191 4 : std::shared_lock<std::shared_mutex> rwLock(baseLock_);
192 4 : auto commIt = channelRemoteRankId_.find(commTag);
193 4 : if (commIt == channelRemoteRankId_.end()) {
194 1 : HCCL_ERROR("[HcclCommDfx]commTag:[%s] not found", commTag.c_str());
195 1 : return HCCL_E_PARA;
196 : }
197 3 : auto handleIt = commIt->second.find(handle);
198 3 : if (handleIt == commIt->second.end()) {
199 1 : HCCL_ERROR("[HcclCommDfx]handle not found,commTag:[%s],handle:[%llu]", commTag.c_str(), handle);
200 1 : return HCCL_E_PARA;
201 : }
202 2 : remoteRankId = handleIt->second;
203 2 : return HCCL_SUCCESS; // 查找成功补充返回成功码
204 4 : }
205 :
206 5 : HcclResult HcclCommDfx::ReportKernel(
207 : uint64_t beginTime, const std::string& commTag, const std::string& kernelName, uint32_t threadId, bool cachedReq)
208 : {
209 5 : CHK_RET(profiling_->ReportKernel(beginTime, commTag, kernelName, threadId, cachedReq));
210 5 : return HCCL_SUCCESS;
211 : }
212 :
213 65548 : u32 HcclCommDfx::GetTaskId(u32 streamId)
214 : {
215 65548 : std::lock_guard<std::mutex> lock(taskIdMutex_);
216 65548 : auto& taskIdRef = streamIdToTaskId_[streamId];
217 65548 : constexpr u32 TASK_ID_MODULO = 65536;
218 65548 : taskIdRef = (taskIdRef + 1) % TASK_ID_MODULO;
219 65548 : u32 retTaskId = taskIdRef;
220 65548 : return retTaskId;
221 65548 : }
222 :
223 3 : void HcclCommDfx::SetDpuStreamId(u32 dpuStreamId) { dpuStreamId_ = dpuStreamId; }
224 :
225 : } // namespace hccl
|
