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 "aicpu_indop_process.h"
12 : #include "coll_comm_aicpu_mgr.h"
13 : #include "hcclCommOp.h"
14 : #include "hcclCommDfxLite.h"
15 : #include <shared_mutex>
16 : #include "env_config/env_config.h"
17 :
18 : using namespace hccl;
19 :
20 : namespace {
21 : struct CollCommAicpuInfo {
22 : std::shared_mutex commAicpuMgrMapMutex; // 读写锁单例，维护全局的读写信息
23 : std::unordered_map<std::string, std::unique_ptr<CollCommAicpuMgr>> commMgrMap;
24 : };
25 : CollCommAicpuInfo g_commAicpuInfo;
26 :
27 : thread_local CollCommAicpuMgr *g_hcclComm = nullptr; // 记录当前线程通信域; AicpuGetCommbyGroup赋值，AicpuReleaseCommMgrbyGroup置空
28 : }
29 :
30 7 : HcclResult AicpuIndopProcess::AicpuIndOpCommInit(CommAicpuParam *commAicpuParam) {
31 7 : CHK_PTR_NULL(commAicpuParam);
32 :
33 6 : CollCommAicpuMgr *commAicpuMgr = nullptr;
34 6 : HcclResult ret = HCCL_SUCCESS;
35 6 : std::string group = commAicpuParam->hcomId;
36 6 : CHK_RET(AcquireAicpuCommMgr(group, &commAicpuMgr));
37 6 : if (commAicpuMgr == nullptr) {
38 0 : HCCL_ERROR("[AicpuIndopProcess][AicpuIndOpCommInit]commAicpu is null group[%s]", group.c_str());
39 0 : return HCCL_E_PTR;
40 : }
41 :
42 6 : ret = commAicpuMgr->InitAicpuIndOp(commAicpuParam);
43 6 : CHK_PRT_RET(ret != HCCL_SUCCESS,
44 : HCCL_ERROR("[AicpuIndopProcess][%s]errNo[0x%016llx] Failed to init independent op comm group[%s]" , __func__,
45 : HCCL_ERROR_CODE(ret), group.c_str()), ret);
46 :
47 6 : return HCCL_SUCCESS;
48 6 : }
49 :
50 6 : HcclResult AicpuIndopProcess::AcquireAicpuCommMgr(const std::string &group, CollCommAicpuMgr **aicpuCommMgrPtr)
51 : {
52 6 : std::unique_lock<std::shared_mutex> rwlock(g_commAicpuInfo.commAicpuMgrMapMutex);
53 : // 查找是否已存在该group的通信实例
54 6 : auto iter = g_commAicpuInfo.commMgrMap.find(group);
55 6 : if (iter != g_commAicpuInfo.commMgrMap.end()) {
56 0 : *aicpuCommMgrPtr = iter->second.get();
57 : // 创建aicpu通信域
58 0 : CHK_RET(iter->second->AcquireCollCommAicpu());
59 0 : HCCL_INFO("[%s]Reuse existing comm group [%s]", __func__, group.c_str());
60 0 : return HCCL_SUCCESS;
61 : }
62 :
63 : // 未找到则创建新实例
64 6 : std::unique_ptr<CollCommAicpuMgr> aicpuCommMgr;
65 6 : EXCEPTION_CATCH(aicpuCommMgr = std::make_unique<CollCommAicpuMgr>(), return HCCL_E_PTR);
66 : // 创建aicpu通信域
67 6 : CHK_RET(aicpuCommMgr->AcquireCollCommAicpu());
68 :
69 : // 将新实例加入映射表
70 6 : *aicpuCommMgrPtr = aicpuCommMgr.get();
71 6 : g_commAicpuInfo.commMgrMap.insert({group, std::move(aicpuCommMgr)});
72 6 : HCCL_RUN_INFO("[%s]Created new comm group [%s]", __func__, group.c_str());
73 6 : return HCCL_SUCCESS;
74 6 : }
75 :
76 2 : HcclResult AicpuIndopProcess::AicpuIndOpThreadInit(ThreadMgrAicpuParam *param)
77 : {
78 2 : CHK_PTR_NULL(param);
79 :
80 1 : std::string group = param->hcomId;
81 1 : HCCL_INFO("[%s]group[%s]", __func__, group.c_str());
82 1 : CollCommAicpuMgr *collCommAicpuMgr = AicpuIndopProcess::AicpuGetCommMgrbyGroup(group);
83 1 : CHK_PRT_RET(collCommAicpuMgr == nullptr, HCCL_ERROR("%s collCommAicpuMgr is null, group[%s]", __func__, group.c_str()), HCCL_E_PTR);
84 1 : HcclResult ret = collCommAicpuMgr->InitThreads(param);
85 1 : CHK_PRT_CONT(ret != HCCL_SUCCESS,
86 : HCCL_ERROR("[AicpuIndopProcess][AicpuIndOpThreadInit]errNo[0x%016llx] Failed to init threads group[%s]",
87 : HCCL_ERROR_CODE(ret), group.c_str()));
88 1 : AicpuReleaseCommMgrbyGroup(group);
89 1 : return ret;
90 1 : }
91 :
92 1 : CollCommAicpuMgr *AicpuIndopProcess::AicpuGetCommMgrbyGroup(const std::string &group)
93 : {
94 1 : HCCL_INFO("[AicpuIndopProcess][%s]start, group[%s]", __func__, group.c_str());
95 1 : auto startTime = std::chrono::steady_clock::now();
96 1 : constexpr u32 pollIntervalUs = 10; // 轮询间隔10us
97 1 : constexpr u32 pollTimeoutMs = 10000; // 等待超过10秒，打印一次日志
98 1 : auto waitPollTimeOutMs = std::chrono::milliseconds(pollTimeoutMs);
99 :
100 : while (true) {
101 1 : std::shared_lock<std::shared_mutex> rwlock(g_commAicpuInfo.commAicpuMgrMapMutex);
102 1 : auto iter = g_commAicpuInfo.commMgrMap.find(group);
103 1 : if (iter == g_commAicpuInfo.commMgrMap.end()) { // 通信域未创建
104 0 : HCCL_ERROR("[AicpuIndopProcess][%s] exist group size is [%u]", __func__, g_commAicpuInfo.commMgrMap.size());
105 0 : auto curIter = g_commAicpuInfo.commMgrMap.begin();
106 :
107 0 : while (curIter != g_commAicpuInfo.commMgrMap.end()) {
108 0 : HCCL_ERROR("[AicpuIndopProcess][%s] exist group [%s]", __func__, curIter->first.c_str());
109 0 : curIter++;
110 : }
111 0 : return nullptr;
112 : }
113 :
114 1 : if (iter->second->IsUsed()) { // 通信域被占用
115 0 : auto curTime = std::chrono::steady_clock::now();
116 0 : if ((curTime - startTime) >= waitPollTimeOutMs) {
117 0 : startTime = curTime;
118 0 : HCCL_RUN_INFO("[AicpuIndopProcess][%s]wait, comm group [%s] has been used", __func__, group.c_str());
119 : }
120 0 : rwlock.unlock();
121 0 : usleep(pollIntervalUs);
122 0 : continue;
123 0 : }
124 1 : g_hcclComm = iter->second.get();
125 1 : iter->second->SetUsed(true);
126 1 : HCCL_INFO("[AicpuIndopProcess][%s]success, group[%s]", __func__, group.c_str());
127 1 : return iter->second.get();
128 1 : }
129 : return nullptr;
130 : }
131 :
132 1 : void AicpuIndopProcess::AicpuReleaseCommMgrbyGroup(const std::string &group)
133 : {
134 1 : std::shared_lock<std::shared_mutex> rwlock(g_commAicpuInfo.commAicpuMgrMapMutex);
135 1 : auto iter = g_commAicpuInfo.commMgrMap.find(group);
136 1 : if (iter == g_commAicpuInfo.commMgrMap.end()) {
137 0 : return;
138 : }
139 1 : g_hcclComm = nullptr;
140 1 : iter->second->SetUsed(false);
141 1 : }
142 :
143 0 : CollCommAicpuMgr *AicpuIndopProcess::AicpuGetComm(const std::string &group)
144 : {
145 0 : if (group.empty()) {
146 0 : HCCL_ERROR("[AicpuIndopProcess][%s] comm group is empty", __func__);
147 0 : return nullptr;
148 : }
149 0 : if (g_hcclComm == nullptr) {
150 0 : HCCL_ERROR("[AicpuIndopProcess][%s] g_hcclComm is nullptr", __func__);
151 0 : return nullptr;
152 : }
153 :
154 0 : if (g_hcclComm->GetCollCommAicpu()->GetIdentifier() != group) {
155 0 : HCCL_ERROR("[AicpuIndopProcess][%s] comm group[%s] is not current comm group", __func__, group.c_str());
156 0 : return nullptr;
157 : }
158 0 : return g_hcclComm;
159 : }
160 :
161 7 : std::shared_mutex& AicpuIndopProcess::AicpuGetCommMutex()
162 : {
163 7 : return g_commAicpuInfo.commAicpuMgrMapMutex;
164 : }
165 :
166 1 : HcclResult AicpuIndopProcess::AicpuIndOpChannelInit(HcclChannelUrmaRes *commParam)
167 : {
168 1 : CHK_PTR_NULL(commParam);
169 :
170 0 : HCCL_INFO("[AicpuIndopProcess][%s] commParam->channelList[%p], commParam->listNum[%u], commParam->uniqueIdAddr[%p], "
171 : "commParam->uniqueIdSize[%u]", __func__, commParam->channelList, commParam->listNum, commParam->uniqueIdAddr,
172 : commParam->uniqueIdSize);
173 :
174 0 : std::string group = commParam->hcomId;
175 0 : CollCommAicpuMgr *collCommAicpuMgr = AicpuIndopProcess::AicpuGetCommMgrbyGroup(group);
176 0 : CHK_PRT_RET(collCommAicpuMgr == nullptr, HCCL_ERROR("%s collCommAicpuMgr is null, group[%s]", __func__, group.c_str()), HCCL_E_PTR);
177 :
178 0 : HcclResult ret = collCommAicpuMgr->AllocChannelResource(commParam);
179 0 : CHK_PRT_CONT(ret != HCCL_SUCCESS,
180 : HCCL_ERROR("[AicpuIndopProcess][AicpuIndOpChannelInit]errNo[0x%016llx] Failed to init channels group[%s]",
181 : HCCL_ERROR_CODE(ret), group.c_str()));
182 :
183 0 : AicpuReleaseCommMgrbyGroup(group);
184 0 : HCCL_INFO("[AicpuIndopProcess][%s] aicpuTask End.", __func__);
185 :
186 0 : return ret;
187 0 : }
188 :
189 1 : HcclResult AicpuIndopProcess::AicpuIndOpChannelUpdate(HcclChannelUrmaRes *commParam)
190 : {
191 1 : CHK_PTR_NULL(commParam);
192 0 : HCCL_INFO("[AicpuIndopProcess][%s] commParam->channelList[%p], commParam->listNum[%u], commParam->uniqueIdAddr[%p], "
193 : "commParam->uniqueIdSize[%u]", __func__, commParam->channelList, commParam->listNum, commParam->uniqueIdAddr,
194 : commParam->uniqueIdSize);
195 :
196 0 : std::string group = commParam->hcomId;
197 0 : CollCommAicpuMgr *collCommAicpuMgr = AicpuIndopProcess::AicpuGetCommMgrbyGroup(group);
198 0 : CHK_PRT_RET(collCommAicpuMgr == nullptr, HCCL_ERROR("%s collCommAicpuMgr is null, group[%s]", __func__, group.c_str()), HCCL_E_PTR);
199 :
200 0 : HcclResult ret = collCommAicpuMgr->UpdateChannelResource(commParam);
201 0 : CHK_PRT_CONT(ret != HCCL_SUCCESS,
202 : HCCL_ERROR("[AicpuIndopProcess][UpdateChannelResource]errNo[0x%016llx] Failed to update channels group[%s]",
203 : HCCL_ERROR_CODE(ret), group.c_str()));
204 :
205 0 : AicpuReleaseCommMgrbyGroup(group);
206 0 : HCCL_INFO("[AicpuIndopProcess][%s] aicpuTask End.", __func__);
207 :
208 0 : return ret;
209 0 : }
210 :
211 1 : HcclResult AicpuIndopProcess::AicpuIndOpNotifyInit(NotifyMgrAicpuParam *param)
212 : {
213 1 : CHK_PTR_NULL(param);
214 :
215 0 : std::string group = param->hcomId;
216 0 : HCCL_INFO("[%s]group[%s]", __func__, group.c_str());
217 0 : CollCommAicpuMgr *collCommAicpuMgr = AicpuIndopProcess::AicpuGetCommMgrbyGroup(group);
218 0 : CHK_PRT_RET(collCommAicpuMgr == nullptr, HCCL_ERROR("%s collCommAicpuMgr is null, group[%s]", __func__, group.c_str()), HCCL_E_PTR);
219 :
220 0 : HcclResult ret = HCCL_E_INTERNAL;
221 0 : if (param->freeFlag) {
222 0 : ret = collCommAicpuMgr->NotifyFree(param);
223 0 : CHK_PRT_CONT(ret != HCCL_SUCCESS,
224 : HCCL_ERROR("[AicpuIndopProcess][%s]errNo[0x%016llx] Failed to free notifys group[%s]",
225 : __func__, HCCL_ERROR_CODE(ret), group.c_str()));
226 : } else {
227 0 : ret = collCommAicpuMgr->NotifyAlloc(param);
228 0 : CHK_PRT_CONT(ret != HCCL_SUCCESS,
229 : HCCL_ERROR("[AicpuIndopProcess][%s]errNo[0x%016llx] Failed to alloc notifys group[%s]",
230 : __func__, HCCL_ERROR_CODE(ret), group.c_str()));
231 : }
232 :
233 0 : HCCL_INFO("[AicpuIndopProcess][%s] comm identifier[%s], notify op[%u] end, num[%u]",
234 : __func__, group.c_str(), param->freeFlag, param->notifyNum);
235 0 : AicpuReleaseCommMgrbyGroup(group);
236 0 : return ret;
237 0 : }
238 :
239 7 : HcclResult AicpuIndopProcess::AicpuGetCommAll(std::vector<std::pair<std::string, CollCommAicpuMgr *>> &aicpuCommInfo)
240 : {
241 12 : for (auto &kv : g_commAicpuInfo.commMgrMap) {
242 5 : aicpuCommInfo.push_back({kv.first, kv.second.get()});
243 : }
244 7 : return HCCL_SUCCESS;
245 : }
246 :
247 4 : HcclResult AicpuIndopProcess::AicpuDestroyCommbyGroup(const std::string &group)
248 : {
249 4 : std::unique_lock<std::shared_mutex> rwlock(g_commAicpuInfo.commAicpuMgrMapMutex);
250 4 : auto iter = g_commAicpuInfo.commMgrMap.find(group);
251 4 : if (iter == g_commAicpuInfo.commMgrMap.end()) {
252 0 : HCCL_ERROR("[AicpuIndopProcess][%s]group[%s] is not exist", __func__, group.c_str());
253 0 : return HCCL_E_PARA;
254 : }
255 :
256 4 : CollCommAicpu* aicpuComm = iter->second->GetCollCommAicpu();
257 4 : CHK_PTR_NULL(aicpuComm);
258 4 : aicpuComm->SetCommmStatus(HcclCommStatus::HCCL_COMM_STATUS_INVALID);
259 :
260 4 : if (iter->second->IsUsed() == true) {
261 0 : HCCL_RUN_WARNING("[AicpuIndopProcess][%s]comm group [%s] has been used, skip erase", __func__, group.c_str());
262 0 : return HCCL_SUCCESS;
263 : }
264 :
265 4 : g_commAicpuInfo.commMgrMap.erase(group);
266 4 : HCCL_RUN_INFO("[AicpuIndopProcess][%s]Destroy comm group [%s] success.", __func__, group.c_str());
267 4 : return HCCL_SUCCESS;
268 4 : }
269 :
270 0 : HcclResult AicpuIndopProcess::AicpuDfxOpInfoInit(HcclDfxOpInfo *aicpuDfxInfo, const std::string& commTag)
271 : {
272 0 : HCCL_INFO("[%s]group[%s], algTag[%s], profiling L0[%d], L1[%d]", __func__, commTag.c_str(), aicpuDfxInfo->algTag,
273 : Hccl::ProfilingHandlerLite::GetInstance().GetProfL0State(),
274 : Hccl::ProfilingHandlerLite::GetInstance().GetProfL1State());
275 :
276 : // 获取device侧的通信域
277 0 : CHK_PRT_RET(g_hcclComm == nullptr, HCCL_ERROR("%s g_hcclComm is null, commTag[%s]", __func__, commTag.c_str()), HCCL_E_PTR);
278 0 : CollCommAicpu* collComm = g_hcclComm->GetCollCommAicpu();
279 0 : CHK_PTR_NULL(collComm);
280 :
281 : // HcclDfxOpInfo 转为DfxOpInfo
282 0 : std::shared_ptr<Hccl::DfxOpInfo> dfxOpInfoOnce = ConvertToDfxOpInfo(*aicpuDfxInfo);
283 0 : dfxOpInfoOnce->opIndex_ = collComm->UpdateIndex();
284 0 : dfxOpInfoOnce->comm_ = reinterpret_cast<void *>(collComm);
285 0 : dfxOpInfoOnce->isIndop_ = true;
286 0 : dfxOpInfoOnce->rankSize_ = collComm->GetTopoInfo().userRankSize;
287 0 : dfxOpInfoOnce->op_.myRank = static_cast<Hccl::RankId>(collComm->GetTopoInfo().userRank);
288 :
289 : // 注册
290 0 : HcclCommDfxLite* hcclCommDfxLite = collComm->GetHcclCommDfxLite();
291 0 : CHK_PTR_NULL(hcclCommDfxLite);
292 0 : CHK_RET(hcclCommDfxLite->SetCurrDfxOpInfo(dfxOpInfoOnce));
293 0 : return HCCL_SUCCESS;
294 0 : }
295 :
296 0 : HcclResult AicpuIndopProcess::ProfilingReportDeviceOp()
297 : {
298 : // 获取device侧的通信域
299 0 : CHK_PTR_NULL(g_hcclComm);
300 0 : CollCommAicpu* collCommAicpu = g_hcclComm->GetCollCommAicpu();
301 0 : CHK_PTR_NULL(collCommAicpu);
302 : // 注册
303 0 : HcclCommDfxLite* hcclCommDfxLite = collCommAicpu->GetHcclCommDfxLite();
304 0 : Hccl::MirrorTaskManagerLite* mirrorTaskMgrLite = hcclCommDfxLite->GetMirrorTaskManagerLite();
305 0 : auto currDfxOpInfo = mirrorTaskMgrLite->GetCurrDfxOpInfo();
306 0 : if (currDfxOpInfo == nullptr) {
307 0 : HCCL_WARNING("[%s] no op info registered , skip ProfilingReportDeviceOp.", __func__);
308 0 : return HCCL_SUCCESS;
309 : }
310 :
311 0 : CHK_RET(hcclCommDfxLite->ReportAllTasks());
312 0 : EXCEPTION_CATCH(Hccl::ProfilingHandlerLite::GetInstance().ReportHcclOpInfo(*currDfxOpInfo),
313 : return HCCL_E_INTERNAL);
314 0 : return HCCL_SUCCESS;
315 0 : }
316 :
317 0 : HcclResult AicpuIndopProcess::UpdateTask(const std::string &group)
318 : {
319 0 : CHK_PTR_NULL(g_hcclComm);
320 0 : CollCommAicpu* collCommAicpu = g_hcclComm->GetCollCommAicpu();
321 0 : CHK_PTR_NULL(collCommAicpu);
322 0 : HcclCommDfxLite* hcclCommDfxLite = collCommAicpu->GetHcclCommDfxLite();
323 0 : CHK_RET(hcclCommDfxLite->UpdateProfStat());
324 0 : return HCCL_SUCCESS;
325 : }
|
