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 "comm_manager.h"
11 :
12 : #include <list>
13 : #include <mutex>
14 : #include <vector>
15 : #include <string>
16 : #include <fstream>
17 : #include <sys/stat.h>
18 : #include <algorithm>
19 : #include <securec.h>
20 : #include <linux/limits.h>
21 : #include <sstream>
22 :
23 : #include "log.h"
24 : #include "hccl/base.h"
25 : #include "hccl_common_v2.h"
26 : #include "hccl/hccl_types.h"
27 : #include "orion_adapter_rts.h"
28 :
29 : #include "tp_manager.h"
30 : #include "inner_net_dev_manager.h"
31 : #include "hccp_hdc_manager.h"
32 : #include "hccp_peer_manager.h"
33 : #include "hccp_tlv_hdc_manager.h"
34 : #include "ccu_driver_handle.h"
35 : #include "rdma_handle_manager.h"
36 : #include "socket_handle_manager.h"
37 : #include "host_socket_handle_manager.h"
38 :
39 : #include "ccu_context_mgr_imp.h"
40 : #include "ccu_res_batch_allocator.h"
41 : #include "ccu_component.h"
42 : #include "communicator_callback.h"
43 : #include "types.h"
44 :
45 : using namespace std;
46 : using namespace Hccl;
47 :
48 : std::mutex g_commInfoV2CtxMutex;
49 :
50 2 : u64 GetFileSize(const std::string& path) {
51 : struct stat fileStat;
52 2 : if (stat(path.c_str(), &fileStat) != 0)
53 : {
54 3 : HCCL_ERROR("[GetFileSize] Get file stat failed , file path:%s", path.c_str());
55 1 : return 0;
56 : }
57 1 : return static_cast<u64>(fileStat.st_size);
58 : }
59 :
60 21 : HcclResult CcuResAllocAndCtxMgrInit(s32 deviceLogicId)
61 : {
62 : try {
63 21 : CcuComponent::GetInstance(deviceLogicId);
64 21 : CcuResBatchAllocator::GetInstance(deviceLogicId);
65 21 : CtxMgrImp::GetInstance(deviceLogicId);
66 0 : } catch (HcclException &e) {
67 0 : HCCL_ERROR(e.what());
68 0 : return e.GetErrorCode();
69 0 : } catch (exception &e) {
70 0 : HCCL_ERROR(e.what());
71 0 : return HcclResult::HCCL_E_INTERNAL;
72 0 : } catch (...) {
73 0 : HCCL_ERROR("Unknown error occurs!");
74 0 : return HcclResult::HCCL_E_INTERNAL;
75 0 : }
76 21 : return HcclResult::HCCL_SUCCESS;
77 : }
78 :
79 : // 规避默认析构顺序导致单例调用接口时序错误，架构优化后必须删除
80 : // 析构含时序要求接口的单例应在此声明
81 : // 声明顺序与期望析构顺序相反
82 23 : HcclResult CallSingletons()
83 : {
84 23 : s32 deviceLogicId = 0;
85 : try {
86 23 : deviceLogicId = HrtGetDevice();
87 : // 避免设备粒度单例访问错误设备
88 22 : if (deviceLogicId < 0 || static_cast<uint32_t>(deviceLogicId) >= ::MAX_MODULE_DEVICE_NUM) {
89 3 : HCCL_WARNING("[CallSingletons] deviceLogicId[%d] may not have device, passed.", deviceLogicId);
90 1 : return HCCL_E_RUNTIME;
91 : }
92 :
93 : // 不同通信域初始化方式时序不同，hdc manager 重复 init 内部会跳过
94 21 : HccpHdcManager::GetInstance();
95 21 : HccpPeerManager::GetInstance(); // host网卡需要拉起peer模式hccp
96 21 : HccpTlvHdcManager::GetInstance();
97 21 : RdmaHandleManager::GetInstance();
98 21 : InnerNetDevManager::GetInstance();
99 21 : SocketHandleManager::GetInstance();
100 21 : HostSocketHandleManager::GetInstance(); // host网卡需要
101 21 : TpManager::GetInstance(deviceLogicId);
102 1 : } catch (HcclException &e) {
103 3 : HCCL_ERROR(e.what());
104 1 : return e.GetErrorCode();
105 1 : } catch (exception &e) {
106 0 : HCCL_ERROR(e.what());
107 0 : return HcclResult::HCCL_E_INTERNAL;
108 0 : } catch (...) {
109 0 : HCCL_ERROR("Unknown error occurs!");
110 0 : return HcclResult::HCCL_E_INTERNAL;
111 0 : }
112 :
113 21 : if (CcuResAllocAndCtxMgrInit(deviceLogicId) != HCCL_SUCCESS) {
114 : // 遗留问题,处理ccu资源申请失败,走aicpu流程
115 0 : HCCL_ERROR("Ccu res batch allocator or ctx mgr init failed.");
116 0 : return HcclResult::HCCL_E_INTERNAL;
117 : }
118 21 : return HcclResult::HCCL_SUCCESS;
119 : }
120 :
121 774 : CommManager &CommManager::GetInstance(s32 deviceLogicId)
122 : {
123 : // 预留额外一个作为兜底通信域
124 840 : static CommManager commManager[::MAX_MODULE_DEVICE_NUM + 1]; // 使用全局命名空间变量
125 :
126 774 : if (deviceLogicId < 0 || static_cast<uint32_t>(deviceLogicId) > ::MAX_MODULE_DEVICE_NUM) {
127 9 : HCCL_WARNING("[GetInstance] deviceLogicId[%d] is invalid, use backup comm instead.", deviceLogicId);
128 3 : deviceLogicId = ::MAX_MODULE_DEVICE_NUM;
129 : }
130 774 : commManager[deviceLogicId].deviceLogicId = deviceLogicId;
131 774 : return commManager[deviceLogicId];
132 : };
133 :
134 485 : HcclCommInfoV2 &CommManager::GetCommInfoV2()
135 : {
136 485 : return commInfoV2;
137 : }
138 :
139 3 : void CommManager::PrintChannelInfo()
140 : {
141 3 : std::lock_guard<std::mutex> lock(commInfoV2.groupParamsLock);
142 3 : u32 channelNum = 0;
143 3 : s32 logicDevId = HrtGetDevice();
144 9 : HCCL_INFO("[CommManager][PrintChannelInfo]devId[%d].", logicDevId);
145 7 : for (u32 dieId = 0; dieId < MAX_CCU_IODIE_NUM; dieId++) {
146 5 : auto ret = CcuGetChannelSpecNum(logicDevId, dieId, channelNum);
147 5 : if (ret != HCCL_SUCCESS) {
148 3 : HCCL_WARNING("[CommManager][PrintChannelInfo]Get channel num failed, devId[%d], dieId[%u]",
149 : logicDevId, dieId);
150 1 : return;
151 : }
152 12 : HCCL_RUN_INFO("[CommManager][PrintChannelInfo]devId[%d], dieId[%u], Channel num[%u].", logicDevId, dieId, channelNum);
153 : }
154 :
155 4 : for (const auto &group : commInfoV2.hcclGroupMap) {
156 6 : for (u32 dieId = 0; dieId < MAX_CCU_IODIE_NUM; dieId++) {
157 4 : u32 channelCount = group.second.pComm->GetUsedChannelCount(dieId);
158 4 : if (channelCount != 0) {
159 6 : HCCL_RUN_INFO("[CommManager][PrintChannelInfo]group[%s], dieId[%u], used channel count[%u].",
160 : group.first.c_str(), dieId, channelCount);
161 : }
162 : }
163 : }
164 3 : }
165 :
166 13 : std::function<void()> CommManager::GetPrintChannelInfoCallback()
167 : {
168 0 : auto callBack = [this]() {
169 0 : PrintChannelInfo();
170 13 : };
171 13 : return callBack;
172 : }
173 :
174 168 : HcclCommInfoV2 &GetCommInfoV2(void)
175 : {
176 168 : std::lock_guard<std::mutex> lock(g_commInfoV2CtxMutex);
177 168 : s32 logicDevId = 0;
178 168 : aclError ret = aclrtGetDevice(&logicDevId);
179 168 : if (ret == ACL_SUCCESS && (static_cast<u32>(logicDevId) < ::MAX_MODULE_DEVICE_NUM)) {
180 : /* 当前线程获取到deviceId, 如果是首次使用该deviceId的HcomInfo, 先判断之前是否已经配置过 */
181 167 : HcclCommInfoV2 &commInfoV2 = CommManager::GetInstance(logicDevId).GetCommInfoV2();
182 167 : if (!commInfoV2.isUsed) {
183 3 : HCCL_WARNING("[GetCommInfoV2] logicDevId[%d] is not Used.", logicDevId);
184 :
185 1 : HcclCommInfoV2 &backupCommInfoV2 = CommManager::GetInstance(::MAX_MODULE_DEVICE_NUM).GetCommInfoV2();
186 1 : if (backupCommInfoV2.isUsed) {
187 0 : return backupCommInfoV2;
188 : }
189 : }
190 167 : commInfoV2.isUsed = true;
191 167 : return commInfoV2;
192 : }
193 :
194 : /* 当前线程没有获取到deviceId, 查找是否有使用过的Ctx */
195 1 : for (u32 i = 0; i <= ::MAX_MODULE_DEVICE_NUM; i++) {
196 1 : HcclCommInfoV2 &commInfoV2 = CommManager::GetInstance(i).GetCommInfoV2();
197 1 : if (commInfoV2.isUsed) {
198 3 : HCCL_WARNING("[GetCommInfoV2] no set device Used logicDevId[%u].", i);
199 1 : return commInfoV2;
200 : }
201 : }
202 :
203 0 : HCCL_WARNING("[GetCommInfoV2] HrtGetDevice fail.");
204 : /* 当前线程没有获取到deviceId, 使用兜底Ctx */
205 0 : HcclCommInfoV2 &backupCommInfoV2 = CommManager::GetInstance(::MAX_MODULE_DEVICE_NUM).GetCommInfoV2();
206 0 : backupCommInfoV2.isUsed = true;
207 0 : return backupCommInfoV2;
208 168 : }
209 :
210 2 : HcclResult GetHcomRankListV2(u32 rankNum, const u32 *rankIds, HcclGroupParamsV2 ¶ms)
211 : {
212 2 : HcclCommInfoV2 &hcomCommInfoV2 = GetCommInfoV2();
213 2 : std::ostringstream printRankIds;
214 :
215 2 : params.totalRanks = rankNum;
216 2 : params.worldRank = hcomCommInfoV2.commParams.myRank;
217 2 : params.groupRank = INVALID_VALUE_RANKID;
218 :
219 2 : unordered_set<uint32_t> rankIdSet;
220 2 : printRankIds << "input rankIds: ";
221 10 : for (u32 i = 0; i < rankNum; i++) {
222 8 : CHK_PTR_NULL(rankIds + i);
223 8 : printRankIds << "rank[";
224 8 : printRankIds << i;
225 8 : printRankIds << "] = ";
226 8 : printRankIds << rankIds[i];
227 8 : if (i < rankNum - 1) {
228 6 : printRankIds << ", ";
229 : }
230 8 : CHK_PRT_RET(
231 : rankIdSet.find(rankIds[i]) != rankIdSet.end(),
232 : HCCL_ERROR("[GetHcomRankListV2]errNo[0x%016llx], " \
233 : "duplicated rankId[%u] in rankIds.",
234 : HCCL_ERROR_CODE(HCCL_E_PARA), rankIds[i]),
235 : HCCL_E_PARA);
236 8 : rankIdSet.insert(rankIds[i]);
237 8 : params.groupRanks.push_back(rankIds[i]);
238 : }
239 6 : HCCL_RUN_INFO("Entry-%s: %s", __func__, printRankIds.str().c_str());
240 :
241 2 : if (params.groupRanks[rankNum - 1] >= hcomCommInfoV2.commParams.rankSize) {
242 0 : HCCL_ERROR("[get][RankList]errNo[0x%016llx] groupRanks[%u]:%u is invalid",
243 : HCOM_ERROR_CODE(HCCL_E_PARA),
244 : rankNum - 1,
245 : params.groupRanks[rankNum - 1]);
246 0 : return HCCL_E_PARA;
247 : }
248 :
249 2 : for (u32 i = 0; i < rankNum; i++) {
250 2 : if (params.groupRanks[i] == params.worldRank) {
251 2 : params.groupRank = i;
252 2 : break;
253 : }
254 : }
255 :
256 2 : u32 serverNum = 1; // severNum初始值应为1，代表groupId为0的serverId;
257 2 : params.serverNum = serverNum;
258 :
259 2 : return HCCL_SUCCESS;
260 2 : }
261 :
262 : // 图模式 创建子通信域 V2
263 4 : HcclResult HcomCreateGroupImplV2(const std::string &group, u32 rankNum, const std::vector<u32> &rankIds)
264 : {
265 4 : HcclUs startut = TIME_NOW();
266 : /* 接口交互信息日志 */
267 4 : rankNum = rankIds.size();
268 4 : std::string rankId = "";
269 20 : for (u32 i = 0; i < rankNum; i++) {
270 16 : rankId += std::to_string(rankIds[i]);
271 16 : if (i < rankNum - 1) {
272 12 : rankId += ',';
273 : }
274 : }
275 12 : HCCL_RUN_INFO("Entry-HcomCreateGroup:group[%s], rankNum[%u], rankIds[%s]", group.c_str(), rankNum, rankId.c_str());
276 :
277 4 : HcclCommInfoV2 &hcomCommInfoV2 = GetCommInfoV2();
278 4 : CHK_PRT_RET(hcomCommInfoV2.pComm == nullptr,
279 : HCCL_ERROR("[Create][Group]hcomCommInfoV2.pComm is null, please check if the initialize process is called."),
280 : HCCL_E_PTR);
281 :
282 : /* 已经存在的group不允许再次创建 */
283 4 : std::unique_lock<std::mutex> groupParaLock(hcomCommInfoV2.groupParamsLock);
284 4 : if (hcomCommInfoV2.hcclGroupMap.find(group) != hcomCommInfoV2.hcclGroupMap.end()) {
285 6 : HCCL_ERROR(
286 : "[Create][Group]errNo[0x%016llx] group[%s] is already exist", HCOM_ERROR_CODE(HCCL_E_PARA), group.c_str());
287 2 : return HCCL_E_PARA;
288 : }
289 2 : groupParaLock.unlock();
290 :
291 : /* 创建groupParamsV2Tem */
292 2 : HcclGroupParamsV2 groupParamsV2Tem;
293 2 : CHK_RET(GetHcomRankListV2(rankNum, rankIds.data(), groupParamsV2Tem));
294 :
295 : /* 如果是groupRank = INVALID_VALUE_RANKID，即本rank不参与create group */
296 2 : if (groupParamsV2Tem.groupRank == INVALID_VALUE_RANKID) {
297 0 : HCCL_ERROR("[Create][Group]errNo[0x%016llx] confirm groupRank from worldRank[%d] error",
298 : HCOM_ERROR_CODE(HCCL_E_NOT_FOUND), hcomCommInfoV2.commParams.myRank);
299 0 : return HCCL_E_NOT_FOUND;
300 : }
301 :
302 : /* 创建子通信域 */
303 2 : Hccl::CommParams subCommParams{group, static_cast<Hccl::RankId>(groupParamsV2Tem.groupRank),
304 2 : rankNum, static_cast<Hccl::RankId>(groupParamsV2Tem.worldRank), hcomCommInfoV2.commParams.devType};
305 2 : auto ret = hcomCommInfoV2.pComm->CreateSubComm(subCommParams, groupParamsV2Tem.groupRanks, groupParamsV2Tem.pComm);
306 2 : CHK_PRT_RET(ret != HCCL_SUCCESS,
307 : HCCL_ERROR("[Create][Group]errNo[0x%016llx] create group failed.", HCOM_ERROR_CODE(ret)), ret);
308 :
309 2 : CHK_SMART_PTR_NULL(groupParamsV2Tem.pComm);
310 2 : groupParamsV2Tem.pComm->RegisterAcceStateCallBack(CommunicatorCallback());
311 2 : s32 logicDevId = HrtGetDevice();
312 2 : CHK_RET(CommManager::GetInstance(logicDevId).SetCommAcceleratorV2(groupParamsV2Tem.pComm.get(), 0)); // 子通信域创建，设置默认accelerator
313 :
314 2 : groupParaLock.lock();
315 2 : hcomCommInfoV2.hcclGroupMap.insert(std::make_pair(group, groupParamsV2Tem));
316 2 : groupParaLock.unlock();
317 :
318 4 : groupParamsV2Tem.pComm->RegisterPrintChannelInfoCallback(
319 4 : CommManager::GetInstance(logicDevId).GetPrintChannelInfoCallback());
320 6 : HCCL_RUN_INFO(
321 : "hcom create group[%s] success, take time [%lld]us", group.c_str(), DURATION_US(TIME_NOW() - startut));
322 :
323 2 : return HCCL_SUCCESS;
324 4 : }
325 :
326 2 : HcclResult HcomDestroyGroupImplV2(const std::string &group)
327 : {
328 2 : HcclCommInfoV2 &hcomCommInfoV2 = GetCommInfoV2();
329 :
330 : /* 接口交互信息日志 */
331 6 : HCCL_RUN_INFO("Entry-HcomDestroyGroup:group[%s]", group.c_str());
332 :
333 2 : std::unique_lock<std::mutex> groupParaLock(hcomCommInfoV2.groupParamsLock);
334 2 : auto iter = hcomCommInfoV2.hcclGroupMap.find(group);
335 2 : if (iter == hcomCommInfoV2.hcclGroupMap.end()) {
336 0 : HCCL_ERROR(
337 : "[Destroy][Group]errNo[0x%016llx] group[%s] is not exist", HCOM_ERROR_CODE(HCCL_E_PARA), group.c_str());
338 0 : return HCCL_E_PARA;
339 : }
340 2 : hcomCommInfoV2.hcclGroupMap.erase(group);
341 : // 通信域销毁，更新ccu使用情况
342 2 : hcomCommInfoV2.ccuStatus.RemoveCommId(group);
343 :
344 2 : groupParaLock.unlock();
345 :
346 6 : HCCL_RUN_INFO("hcom destroy group[%s] success.", group.c_str());
347 2 : return HCCL_SUCCESS;
348 2 : }
349 :
350 1 : HcclResult HcomGetWorldRankFromGroupRankV2(const char *group, u32 groupRank, u32 *worldRank)
351 : {
352 1 : HcclCommInfoV2 &hcomCommInfoV2 = GetCommInfoV2();
353 : // 校验通信域非空
354 1 : CHK_PRT_RET(hcomCommInfoV2.pComm == nullptr,
355 : HCCL_ERROR("[Get][WorldRank]hcomCommInfoV2.pComm is null, "
356 : "please check if the initialize process is called."),
357 : HCCL_E_PTR);
358 : // 获取group
359 1 : std::string strGroup = (group == nullptr) ? HCCL_WORLD_GROUP : group;
360 1 : if (strGroup == HCCL_WORLD_GROUP) {
361 1 : *worldRank = hcomCommInfoV2.commParams.myRank;
362 3 : HCCL_INFO("hcom get world rank success, group[%s], groupRank[%u], worldRank[%u]",
363 : strGroup.c_str(), groupRank, *worldRank);
364 1 : return HCCL_SUCCESS;
365 : }
366 0 : std::unique_lock<std::mutex> groupParaLock(hcomCommInfoV2.groupParamsLock);
367 0 : auto iter = hcomCommInfoV2.hcclGroupMap.find(strGroup);
368 0 : if (iter == hcomCommInfoV2.hcclGroupMap.end()) {
369 0 : HCCL_ERROR(
370 : "[Get][WorldRank]errNo[0x%016llx] group[%s] is not exist", HCOM_ERROR_CODE(HCCL_E_PARA), strGroup.c_str());
371 0 : return HCCL_E_PARA;
372 : }
373 :
374 : // groupRanks判空
375 0 : CHK_PRT_RET((iter->second).groupRanks.empty(),
376 : HCCL_ERROR("[Get][WorldRank]errNo[0x%016llx] group[%s] ranks is empty",
377 : HCOM_ERROR_CODE(HCCL_E_INTERNAL), strGroup.c_str()), HCCL_E_INTERNAL);
378 :
379 : // 校验groupRank合法性
380 0 : if (groupRank >= (iter->second).totalRanks) {
381 0 : HCCL_ERROR("[Get][WorldRank]errNo[0x%016llx] group[%s] groupRank[%u] is invalid",
382 : HCOM_ERROR_CODE(HCCL_E_PARA), strGroup.c_str(), groupRank);
383 0 : return HCCL_E_PARA;
384 : }
385 0 : *worldRank = (iter->second).groupRanks[groupRank];
386 :
387 0 : HCCL_INFO("hcom get world rank success, group[%s], groupRank[%u], worldRank[%u]",
388 : strGroup.c_str(), groupRank, *worldRank);
389 0 : return HCCL_SUCCESS;
390 1 : }
391 :
392 1 : HcclResult HcomGetGroupRankFromWorldRankV2(u32 worldRank, const char *group, u32 *groupRank)
393 : {
394 1 : HcclCommInfoV2 &hcomCommInfoV2 = GetCommInfoV2();
395 : // 校验通信域非空
396 1 : CHK_PRT_RET(hcomCommInfoV2.pComm == nullptr,
397 : HCCL_ERROR("[Get][GroupRank]hcomCommInfoV2.pComm is null, "
398 : "please check if the initialize process is called."),
399 : HCCL_E_PTR);
400 : // 校验worldRank合法性
401 1 : if (worldRank >= hcomCommInfoV2.commParams.rankSize) {
402 0 : HCCL_ERROR(
403 : "[Get][GroupRank]errNo[0x%016llx] world[%u] rank is invalid", HCOM_ERROR_CODE(HCCL_E_PARA), worldRank);
404 0 : return HCCL_E_PARA;
405 : }
406 : // 获取group
407 1 : std::string strGroup = (group == nullptr) ? HCCL_WORLD_GROUP : group;
408 1 : if (strGroup == HCCL_WORLD_GROUP) {
409 1 : *groupRank = hcomCommInfoV2.commParams.myRank;
410 3 : HCCL_INFO("hcom get group rank success, group[%s], worldRank[%u], groupRank[%u]",
411 : strGroup.c_str(), worldRank, *groupRank);
412 1 : return HCCL_SUCCESS;
413 : }
414 0 : std::unique_lock<std::mutex> groupParaLock(hcomCommInfoV2.groupParamsLock);
415 0 : auto iter = hcomCommInfoV2.hcclGroupMap.find(strGroup);
416 0 : if (iter == hcomCommInfoV2.hcclGroupMap.end()) {
417 0 : HCCL_ERROR(
418 : "[Get][GroupRank]errNo[0x%016llx] group[%s] is not exist", HCOM_ERROR_CODE(HCCL_E_PARA), strGroup.c_str());
419 0 : return HCCL_E_PARA;
420 : }
421 :
422 : // groupRanks判空
423 0 : CHK_PRT_RET((iter->second).groupRanks.empty(),
424 : HCCL_ERROR("[Get][GroupRank]errNo[0x%016llx] group[%s] ranks is empty",
425 : HCOM_ERROR_CODE(HCCL_E_INTERNAL), strGroup.c_str()), HCCL_E_INTERNAL);
426 :
427 : // 获取groupRank
428 0 : for (u32 rank = 0; rank < (iter->second).totalRanks; rank++) {
429 0 : if (worldRank == (iter->second).groupRanks[rank]) {
430 0 : *groupRank = rank;
431 :
432 0 : HCCL_INFO("hcom get group rank success, group[%s], worldRank[%u], groupRank[%u]",
433 : strGroup.c_str(), worldRank, *groupRank);
434 0 : return HCCL_SUCCESS;
435 : }
436 : }
437 0 : return HCCL_E_PARA;
438 1 : }
439 :
440 1 : HcclResult HcomGetRankSizeV2(const char *group, u32 *rankSize)
441 : {
442 1 : HcclCommInfoV2 &hcomCommInfoV2 = GetCommInfoV2();
443 : // 校验通信域非空
444 1 : CHK_PRT_RET(hcomCommInfoV2.pComm == nullptr,
445 : HCCL_ERROR("[Get][RankSize]hcomCommInfoV2.pComm is null, "
446 : "please check if the initialize process is called."),
447 : HCCL_E_PTR);
448 : // 获取group
449 1 : std::string strGroup = (group == nullptr) ? HCCL_WORLD_GROUP : group;
450 1 : if (strGroup == HCCL_WORLD_GROUP) {
451 1 : *rankSize = hcomCommInfoV2.commParams.rankSize;
452 3 : HCCL_INFO("hcom get world rank size success, rankSize[%u]", *rankSize);
453 1 : return HCCL_SUCCESS;
454 : }
455 0 : std::unique_lock<std::mutex> groupParaLock(hcomCommInfoV2.groupParamsLock);
456 0 : auto iter = hcomCommInfoV2.hcclGroupMap.find(strGroup);
457 0 : if (iter == hcomCommInfoV2.hcclGroupMap.end()) {
458 0 : HCCL_ERROR(
459 : "[Get][RankSize]errNo[0x%016llx] group[%s] is not exist", HCOM_ERROR_CODE(HCCL_E_PARA), strGroup.c_str());
460 0 : return HCCL_E_PARA;
461 : }
462 0 : CHK_SMART_PTR_NULL((iter->second).pComm);
463 0 : HcclResult ret = (iter->second).pComm->GetRankSize(rankSize);
464 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[Get][RankSize]GetRankSize failed."), HCCL_E_PTR);
465 0 : groupParaLock.unlock();
466 :
467 0 : HCCL_INFO("hcom get rank size success, group[%s], rankSize[%u]", strGroup.c_str(), *rankSize);
468 0 : return HCCL_SUCCESS;
469 1 : }
470 :
471 1 : HcclResult HcomGetCommV2(void **commV2)
472 : {
473 1 : CHK_PTR_NULL(commV2);
474 1 : HcclCommInfoV2 &hcomCommInfoV2 = GetCommInfoV2();
475 1 : CHK_PTR_NULL(hcomCommInfoV2.pComm);
476 1 : *commV2 = static_cast<void *>(hcomCommInfoV2.pComm.get());
477 3 : HCCL_INFO("[HcomGetCommV2] success.");
478 1 : return HCCL_SUCCESS;
479 : }
480 :
481 1 : HcclResult HcomGetGroupParamsV2(const char *group, void* groupParams, void **commV2)
482 : {
483 1 : HcclCommInfoV2 &hcomCommInfoV2 = GetCommInfoV2();
484 2 : auto iter = hcomCommInfoV2.hcclGroupMap.find(group);
485 1 : if (iter == hcomCommInfoV2.hcclGroupMap.end()) {
486 0 : HCCL_ERROR("[HcomGetGroupParamsV2] group[%s] not found", group);
487 0 : return HCCL_E_PARA;
488 : }
489 1 : HcclGroupParamsV2 &groupParamsV2 = iter->second;
490 1 : HcclGroupParamsV2 *groupParamsTem = static_cast<HcclGroupParamsV2*>(groupParams);
491 1 : *groupParamsTem = groupParamsV2;
492 1 : CHK_PTR_NULL(groupParamsV2.pComm);
493 1 : *commV2 = static_cast<Hccl::HcclCommunicator*>(groupParamsV2.pComm.get());
494 3 : HCCL_INFO("[HcomGetGroupParamsV2] success. group[%s]", group);
495 1 : return HCCL_SUCCESS;
496 : }
497 :
498 1 : HcclResult HcomDestroyV2(void)
499 : {
500 1 : HcclCommInfoV2 &hcomCommInfoV2 = GetCommInfoV2();
501 1 : if (hcomCommInfoV2.pComm != nullptr) {
502 1 : bool isSetDeviceByHcomm = false;
503 1 : if (hcomCommInfoV2.devId != HOST_DEVICE_ID) {
504 1 : s32 logicDevId = 0;
505 1 : aclError ret = aclrtGetDevice(&logicDevId);
506 1 : if (ret == ACL_ERROR_RT_CONTEXT_NULL) {
507 : // 若当前线程获取不到context，则由hccl进行setDevice，并在通信域析构完成后resetDevice
508 0 : HrtSetDevice(hcomCommInfoV2.devId);
509 0 : isSetDeviceByHcomm = true;
510 1 : } else if (ret != ACL_SUCCESS) {
511 0 : HCCL_ERROR("[HcomDestroyV2] get device failed, ret[%d]", ret);
512 : }
513 : }
514 : // 通信域销毁，更新ccu使用情况
515 1 : hcomCommInfoV2.ccuStatus.RemoveCommId(hcomCommInfoV2.pComm->GetId());
516 1 : hcomCommInfoV2.pComm = nullptr;
517 1 : std::unique_lock<std::mutex> groupParaLock(hcomCommInfoV2.groupParamsLock);
518 :
519 : // 通信域销毁，更新子通信域ccu使用情况
520 2 : for (auto iterGroup : hcomCommInfoV2.hcclGroupMap) {
521 1 : hcomCommInfoV2.ccuStatus.RemoveCommId(iterGroup.first);
522 1 : }
523 1 : hcomCommInfoV2.hcclGroupMap.clear();
524 1 : if (isSetDeviceByHcomm) {
525 0 : HrtResetDevice(hcomCommInfoV2.devId);
526 : }
527 1 : }
528 1 : return HCCL_SUCCESS;
529 : }
530 :
531 1 : static HcclResult GetRankTableInfo(const char *rankTablePath, std::string &ranktableInfo)
532 : {
533 : // 校验文件是否存在
534 1 : char resolvedPath[PATH_MAX] = {0};
535 1 : if (realpath(rankTablePath, resolvedPath) == nullptr) {
536 0 : HCCL_ERROR("RanktableRealPath: %s is not a valid real path", rankTablePath);
537 0 : return HCCL_E_INTERNAL;
538 : }
539 :
540 3 : HCCL_INFO("waiting for json file load complete");
541 1 : u64 ranktableFileSize = GetFileSize(resolvedPath);
542 1 : if (ranktableFileSize > RANKTABLE_FILE_MAX_SIZE || ranktableFileSize <= 0) {
543 3 : HCCL_ERROR("[GetRankTableInfo] ranktablefile size: %u, ranktable must be greater than 0 and less than %u", ranktableFileSize, RANKTABLE_FILE_MAX_SIZE);
544 1 : return HCCL_E_OPEN_FILE_FAILURE;
545 : }
546 :
547 0 : std::ifstream infoFile(resolvedPath, std::ifstream::in);
548 0 : if (!infoFile) {
549 0 : HCCL_ERROR("open file %s failed", resolvedPath);
550 0 : return HCCL_E_INTERNAL;
551 : }
552 :
553 0 : std::stringstream rankTableStr;
554 0 : rankTableStr << infoFile.rdbuf();
555 0 : ranktableInfo = rankTableStr.str();
556 :
557 0 : return HCCL_SUCCESS;
558 0 : }
559 :
560 : // 图模式 创建全局通信域 V2
561 1 : HcclResult HcomInitByFileV2(const char *rankTablePath, const char *identify)
562 : {
563 : // 待解决：目前主要为了芯片验证，非最终版本
564 3 : HCCL_RUN_INFO("Entry-HcomInitByFile V950, ranktable[%s], identify[%s]", rankTablePath, identify);
565 :
566 : // 解析myRank
567 : s32 myRank;
568 : try {
569 1 : myRank = std::atoi(identify);
570 : } catch (...) {
571 : HCCL_ERROR("atoi(identify) failed!");
572 : return HCCL_E_INTERNAL;
573 : }
574 :
575 1 : CallSingletons(); // 临时规避，在初始化通信域前声明单例保证时序
576 :
577 : // 防止重复调用初始化
578 1 : string commId(HCCL_WORLD_GROUP);
579 1 : HcclCommInfoV2 &hcomCommInfoV2 = GetCommInfoV2();
580 1 : CHK_PRT_RET(hcomCommInfoV2.hcclGroupMap.find(commId) != hcomCommInfoV2.hcclGroupMap.end(),
581 : HCCL_ERROR("[Init][CheckOpBasedHcom]errNo[0x%016llx] The comm name[%s] already exists in Group2Comm map.",
582 : HCCL_ERROR_CODE(HCCL_E_PARA), commId.c_str()), HCCL_E_PARA);
583 :
584 : // 解析ranktable
585 1 : std::string ranktableInfo;
586 1 : HcclResult ret = GetRankTableInfo(rankTablePath, ranktableInfo);
587 4 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[HcomInitByFile] get ranktable info failed"), ret);
588 :
589 0 : bool devUsed = false;
590 0 : bool isWorldGroup = true;
591 : // 临时修改这个为4，后边要改掉这个，在初始流程中，解析完虚拟拓扑后，添加ranksize
592 : Hccl::CommParams commParams{commId, static_cast<Hccl::RankId>(myRank), 0, static_cast<Hccl::RankId>(myRank),
593 0 : Hccl::HrtGetDeviceType(), devUsed, isWorldGroup};
594 0 : hcomCommInfoV2.pComm.reset(new (std::nothrow) Hccl::HcclCommunicator(commParams));
595 0 : CHK_PTR_NULL(hcomCommInfoV2.pComm);
596 0 : auto res = hcomCommInfoV2.pComm->Init(ranktableInfo);
597 0 : CHK_PRT_RET(res != HcclResult::HCCL_SUCCESS,
598 : HCCL_ERROR("[HcomInitByFile] Hccl::Communicator Init failed, res %d", res), HCCL_E_INTERNAL);
599 :
600 0 : hcomCommInfoV2.pComm->RegisterAcceStateCallBack(CommunicatorCallback());
601 0 : s32 logicDevId = HrtGetDevice();
602 0 : CHK_RET(CommManager::GetInstance(logicDevId).SetCommAcceleratorV2(hcomCommInfoV2.pComm.get(), 0)); // 全局通信域创建，设置默认accelerator
603 :
604 0 : res = hcomCommInfoV2.pComm->GetRankSize(&commParams.rankSize);
605 0 : CHK_PRT_RET(res != HCCL_SUCCESS,
606 : HCCL_ERROR("[HcomInitByFile] Hccl::Communicator GetRankSize failed, rankSize = %u", commParams.rankSize), res);
607 0 : hcomCommInfoV2.commParams = commParams;
608 :
609 0 : HcclGroupParamsV2 params;
610 0 : params.pComm = hcomCommInfoV2.pComm;
611 0 : std::unique_lock<std::mutex> groupParaLock(hcomCommInfoV2.groupParamsLock);
612 0 : hcomCommInfoV2.hcclGroupMap[commId] = params;
613 0 : groupParaLock.unlock();
614 :
615 0 : hcomCommInfoV2.pComm->RegisterPrintChannelInfoCallback(
616 0 : CommManager::GetInstance(logicDevId).GetPrintChannelInfoCallback());
617 :
618 0 : HCCL_INFO(
619 : "[HcomInitByFile] HcomInitByFile success! logicDevId[%d], commId[%s]", logicDevId, commParams.commId.c_str());
620 :
621 0 : return HCCL_SUCCESS;
622 1 : }
623 :
624 1 : HcclResult HcomInitByStringV2(const char *rankTableM, const char *identify)
625 : {
626 : // 待解决：目前主要为了芯片验证，非最终版本
627 3 : HCCL_RUN_INFO("Entry-HcomInitByString V950, rankTableM[%s], identify[%s]", rankTableM, identify);
628 :
629 : // 解析myRank
630 : s32 myRank;
631 : try {
632 1 : myRank = std::atoi(identify);
633 : } catch (...) {
634 : HCCL_ERROR("atoi(identify) failed!");
635 : return HCCL_E_INTERNAL;
636 : }
637 :
638 1 : CallSingletons(); // 临时规避，在初始化通信域前声明单例保证时序
639 :
640 : // 防止重复调用初始化
641 1 : string commId(HCCL_WORLD_GROUP);
642 1 : HcclCommInfoV2 &hcomCommInfoV2 = GetCommInfoV2();
643 1 : CHK_PRT_RET(hcomCommInfoV2.hcclGroupMap.find(commId) != hcomCommInfoV2.hcclGroupMap.end(),
644 : HCCL_ERROR("[Init][CheckOpBasedHcom]errNo[0x%016llx] The comm name[%s] already exists in Group2Comm map.",
645 : HCCL_ERROR_CODE(HCCL_E_PARA), commId.c_str()), HCCL_E_PARA);
646 :
647 1 : bool devUsed = false;
648 1 : bool isWorldGroup = true;
649 : // 临时修改这个为4，后边要改掉这个，在初始流程中，解析完虚拟拓扑后，添加ranksize
650 : Hccl::CommParams commParams{commId, static_cast<Hccl::RankId>(myRank), 0, static_cast<Hccl::RankId>(myRank),
651 1 : Hccl::HrtGetDeviceType(), devUsed, isWorldGroup};
652 1 : hcomCommInfoV2.pComm.reset(new (std::nothrow) Hccl::HcclCommunicator(commParams));
653 1 : CHK_PTR_NULL(hcomCommInfoV2.pComm);
654 2 : auto res = hcomCommInfoV2.pComm->Init(rankTableM);
655 1 : CHK_PRT_RET(res != HcclResult::HCCL_SUCCESS,
656 : HCCL_ERROR("[HcomInitByString] Hccl::Communicator Init failed, res %d", res), HCCL_E_INTERNAL);
657 :
658 1 : hcomCommInfoV2.pComm->RegisterAcceStateCallBack(CommunicatorCallback());
659 1 : s32 logicDevId = HrtGetDevice();
660 1 : hcomCommInfoV2.devId = logicDevId;
661 1 : CHK_RET(CommManager::GetInstance(logicDevId).SetCommAcceleratorV2(hcomCommInfoV2.pComm.get(), 0)); // 全局通信域创建，设置默认accelerator
662 :
663 1 : res = hcomCommInfoV2.pComm->GetRankSize(&commParams.rankSize);
664 1 : CHK_PRT_RET(res != HCCL_SUCCESS,
665 : HCCL_ERROR("[HcomInitByString] Hccl::Communicator GetRankSize failed, rankSize = %u", commParams.rankSize), res);
666 1 : hcomCommInfoV2.commParams = commParams;
667 :
668 1 : HcclGroupParamsV2 params;
669 1 : params.pComm = hcomCommInfoV2.pComm;
670 1 : std::unique_lock<std::mutex> groupParaLock(hcomCommInfoV2.groupParamsLock);
671 1 : hcomCommInfoV2.hcclGroupMap[commId] = params;
672 1 : groupParaLock.unlock();
673 :
674 2 : hcomCommInfoV2.pComm->RegisterPrintChannelInfoCallback(
675 2 : CommManager::GetInstance(logicDevId).GetPrintChannelInfoCallback());
676 :
677 3 : HCCL_INFO(
678 : "[HcomInitByString] HcomInitByString success! logicDevId[%d], commId[%s]", logicDevId, commParams.commId.c_str());
679 :
680 1 : return HCCL_SUCCESS;
681 1 : }
682 :
683 :
684 17 : void CcuStatus::RemoveCommId(const std::string &commId)
685 : {
686 17 : auto itMs = std::find(useMsCommIds.begin(), useMsCommIds.end(), commId);
687 17 : if (itMs != useMsCommIds.end()) {
688 0 : HCCL_DEBUG("[CcuStatus][%s] commId[%s] used ccu ms, removed", __func__, commId.c_str());
689 0 : useMsCommIds.erase(itMs);
690 : }
691 :
692 17 : auto itSched = std::find(useSchedCommIds.begin(), useSchedCommIds.end(), commId);
693 17 : if (itSched != useSchedCommIds.end()) {
694 3 : HCCL_DEBUG("[CcuStatus][%s] commId[%s] used ccu sched, removed", __func__, commId.c_str());
695 1 : useSchedCommIds.erase(itSched);
696 : }
697 17 : }
698 :
699 16 : bool CcuStatus::IsMsAvailable(const std::string &commId) const
700 : {
701 16 : auto itMs = std::find(useMsCommIds.begin(), useMsCommIds.end(), commId);
702 : // ms没有通信域使用，或者就是传入通信域在使用，则可用
703 16 : return (useMsCommIds.size() < MAX_NUM_COMM_USING_MS) || (itMs != useMsCommIds.end());
704 : }
705 :
706 6 : HcclResult CcuStatus::InsertCommId(const std::string &commId, bool isUsingCcuMs, bool isUsingCcuSched)
707 : {
708 : // 先删再加，避免重复添加到两种模式
709 6 : RemoveCommId(commId);
710 : // ccu ms 没有被使用过，则将ccu ms 标记为已使用
711 6 : if (isUsingCcuMs) {
712 5 : CHK_RET(InsertMsCommId(commId));
713 4 : } else if (isUsingCcuSched) {
714 2 : InsertSchedCommId(commId);
715 : } else {
716 6 : HCCL_DEBUG("NotUsingCcu comm [%s]", commId.c_str());
717 : }
718 5 : return HCCL_SUCCESS;
719 : }
720 :
721 2 : HcclResult CcuStatus::InsertMsCommId(const std::string &commId)
722 : {
723 2 : if (!IsMsAvailable(commId)) {
724 3 : HCCL_WARNING("[%s] ccu ms has been used by comm [%s], no more than 2 comms can use ccu ms at the same time.",
725 : __func__, (*(useMsCommIds.begin())).c_str());
726 1 : return HCCL_E_INTERNAL;
727 : }
728 3 : HCCL_DEBUG("[%s] UsingCcuMs comm [%s]", __func__, commId.c_str());
729 1 : useMsCommIds.push_back(commId);
730 1 : return HCCL_SUCCESS;
731 : }
732 :
733 2 : void CcuStatus::InsertSchedCommId(const std::string &commId)
734 : {
735 6 : HCCL_DEBUG("[%s] UsingCcuSched comm [%s]", __func__, commId.c_str());
736 2 : useSchedCommIds.push_back(commId);
737 2 : }
738 :
739 14 : HcclResult CommManager::SetCommAcceleratorV2(Hccl::HcclCommunicator *communicator, int32_t accelerator)
740 : {
741 14 : CHK_PTR_NULL(communicator);
742 14 : if (accelerator < static_cast<int32_t>(HcclAccelerator::DEFAULT) || accelerator > static_cast<int32_t>(HcclAccelerator::AICPU)) {
743 0 : HCCL_ERROR("[SetCommAcceleratorV2] Invalid accelerator value [%d], valid range is [0,7]", accelerator);
744 0 : return HCCL_E_NOT_SUPPORT;
745 : }
746 14 : HcclAccelerator hcclAccelerator = static_cast<HcclAccelerator::Value>(accelerator);
747 :
748 14 : HcclCommInfoV2 &opbasedCommInfoV2 = GetCommInfoV2();
749 : // 通过进程锁看护，避免多个通信域同时占用CCU_MS
750 14 : std::unique_lock<std::mutex> lock(opbasedCommInfoV2.groupParamsLock);
751 14 : if ((hcclAccelerator == HcclAccelerator::CCU_MS || hcclAccelerator == HcclAccelerator::CCU_SCHED) && !isCcuAvailable) {
752 0 : HCCL_WARNING("CCU not support reuse in single device multi-precess services, accelerator fallback AICPU_TS");
753 0 : hcclAccelerator = HcclAccelerator::AICPU_TS;
754 : }
755 14 : bool isMsAvailable = opbasedCommInfoV2.ccuStatus.IsMsAvailable(communicator->GetId());
756 42 : HCCL_INFO("[CommManager][%s] hcclAccelerator is [%s], isMsAvailable is [%d]", __func__, hcclAccelerator.Describe().c_str(),
757 : isMsAvailable);
758 14 : CHK_RET(communicator->SetAccelerator(hcclAccelerator, isMsAvailable));
759 14 : return HCCL_SUCCESS;
760 14 : }
761 :
762 0 : std::shared_ptr<Hccl::CcuDriverHandle> CommManager::GetCcuDriver()
763 : {
764 0 : if (isCcuAvailable == true && ccuDriverHandle == nullptr) {
765 0 : ccuDriverHandle = std::make_shared<Hccl::CcuDriverHandle>(deviceLogicId);
766 0 : if (ccuDriverHandle->Init() == HCCL_E_UNAVAIL) {
767 0 : isCcuAvailable = false;
768 0 : ccuDriverHandle = nullptr;
769 0 : HCCL_WARNING("[CommManager::GetCcuDriver]Tlv already open, isCcuAvailable updated to false");
770 : }
771 : }
772 0 : return ccuDriverHandle;
773 : }
774 :
775 270 : void CommManager::DeinitCcuDriver() {
776 270 : if (ccuDriverHandle.use_count() == 1) {
777 0 : ccuDriverHandle = nullptr;
778 : }
779 270 : }
780 :
781 3 : HcclResult HcomGetCcuTaskInfo(const std::string &group, void *tilingData, void *ccuTaskGroup)
782 : {
783 3 : CHK_PTR_NULL(tilingData);
784 3 : CHK_PTR_NULL(ccuTaskGroup);
785 3 : CHK_PRT_RET(group.empty(), HCCL_ERROR("[HcomGetCcuTaskInfo] group is null"), HCCL_E_PARA);
786 :
787 : /* 接口交互信息日志 */
788 9 : HCCL_RUN_INFO("HcomGetCcuTaskInfo:group[%s]", group.c_str());
789 :
790 3 : HcclCommInfoV2 &hcomCommInfoV2 = GetCommInfoV2();
791 :
792 3 : std::unique_lock<std::mutex> groupParaLock(hcomCommInfoV2.groupParamsLock);
793 3 : auto iter = hcomCommInfoV2.hcclGroupMap.find(group);
794 3 : if (iter == hcomCommInfoV2.hcclGroupMap.end()) {
795 3 : HCCL_ERROR(
796 : "[HcomGetCcuTaskInfo]errNo[0x%016llx] group[%s] is not exist", HCOM_ERROR_CODE(HCCL_E_PARA), group.c_str());
797 1 : return HCCL_E_PARA;
798 : }
799 2 : HcclGroupParamsV2 &groupParam = iter->second;
800 2 : Hccl::HcclCommunicator *comm = static_cast<Hccl::HcclCommunicator *>(groupParam.pComm.get());
801 2 : CHK_PTR_NULL(comm);
802 2 : auto ret = comm->GetCcuTaskInfo(tilingData, ccuTaskGroup);
803 2 : if (ret != HCCL_SUCCESS) {
804 3 : HCCL_ERROR("[HcomGetCcuTaskInfo] GetCcuTaskInfo failed.");
805 1 : return HCCL_E_INTERNAL;
806 : }
807 :
808 3 : HCCL_RUN_INFO("HcomGetCcuTaskInfo success group[%s]", group.c_str());
809 1 : return HCCL_SUCCESS;
810 3 : }
|
