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