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