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 "my_rank.h"
11 : #include "hcomm_c_adpt.h"
12 : #include "endpoint_pair.h"
13 : #include "hccl_res.h"
14 : #include "../common/loggers/channel_logger.h" // 日志记录器
15 : #include "hcclCommDfx.h"
16 : #include "config/env_config.h"
17 : #include "env_config/env_config.h"
18 : #include "channel_process.h"
19 : #include "ccu_dev_mgr_imp.h"
20 : #include "ccu_device_res.h"
21 : #include "ccu_res_desc.h"
22 : #include "ccu_device_pub.h"
23 : #include "ccu_res_desc_mgr.h"
24 : #include "ccu_log.h"
25 : #include "dlprof_function.h"
26 : #include "config_log.h"
27 : #include "comm_engine_utils.h"
28 : #include "hcom_common.h"
29 : #include "op_base.h"
30 : #include "ccu_res.h"
31 : #include "coll_comm_mgr.h"
32 :
33 : #include <acl/acl.h>
34 :
35 : using namespace hcomm;
36 :
37 : namespace MyRankUtils {
38 :
39 4 : uint32_t ResolveUbCommDomainQos(const hccl::CommConfig &commConfig)
40 : {
41 4 : if (commConfig.GetConfigHcclQos() == HCCL_COMM_QOS_CONFIG_NOT_SET) {
42 1 : return EnvConfig::UB_QOS_DEFAULT;
43 : }
44 3 : return commConfig.GetConfigHcclQos();
45 : }
46 :
47 13 : HcommChannelDesc ChannelDescHccl2Hcomm(const HcclChannelDesc &hcclDesc, const hccl::CommConfig &commConfig)
48 : {
49 13 : HcommChannelDesc hcommDesc{};
50 13 : (void)HcommChannelDescInit(&hcommDesc, 1);
51 13 : hcommDesc.remoteEndpoint = hcclDesc.remoteEndpoint;
52 13 : hcommDesc.notifyNum = hcclDesc.notifyNum;
53 13 : hcommDesc.memHandles = reinterpret_cast<HcommMemHandle *>(hcclDesc.memHandles);
54 13 : hcommDesc.memHandleNum = hcclDesc.memHandleNum;
55 13 : (void)memcpy_s(hcommDesc.raws, sizeof(hcommDesc.raws), hcclDesc.raws, sizeof(hcommDesc.raws));
56 13 : if (hcclDesc.channelProtocol == COMM_PROTOCOL_ROCE) {
57 5 : hcommDesc.roceAttr.retryCnt = hcclDesc.roceAttr.retryCnt;
58 5 : hcommDesc.roceAttr.retryInterval = hcclDesc.roceAttr.retryInterval;
59 5 : hcommDesc.roceAttr.sl = hcclDesc.roceAttr.sl;
60 5 : hcommDesc.roceAttr.tc = hcclDesc.roceAttr.tc;
61 5 : return hcommDesc;
62 : }
63 8 : if (hcclDesc.channelProtocol == COMM_PROTOCOL_UBC_CTP ||
64 7 : hcclDesc.channelProtocol == COMM_PROTOCOL_UBC_TP ||
65 5 : hcclDesc.channelProtocol == COMM_PROTOCOL_UBOE ||
66 4 : hcclDesc.channelProtocol == COMM_PROTOCOL_UBG) {
67 4 : hcommDesc.qos = ResolveUbCommDomainQos(commConfig);
68 : }
69 8 : return hcommDesc;
70 : }
71 :
72 : /* 公共模块函数返回值定义，跟业务层同步 */
73 : const std::unordered_map<CommProtocol, std::string> HCOM_COMM_PROTOCOL_STR_MAP = {
74 : {COMM_PROTOCOL_RESERVED, "RESERVED"},
75 : {COMM_PROTOCOL_HCCS, "HCCS"},
76 : {COMM_PROTOCOL_ROCE, "ROCE"},
77 : {COMM_PROTOCOL_PCIE, "PCIE"},
78 : {COMM_PROTOCOL_SIO, "SIO"},
79 : {COMM_PROTOCOL_UBC_CTP, "UBC_CTP"},
80 : {COMM_PROTOCOL_UBC_TP, "UBC_TP"},
81 : {COMM_PROTOCOL_UB_MEM, "UB_MEM"},
82 : {COMM_PROTOCOL_UBOE, "UBOE"},
83 : {COMM_PROTOCOL_UBG, "UBG"}
84 : };
85 :
86 0 : inline std::string GetCommProtocolEnumStr(CommProtocol protocol)
87 : {
88 0 : auto iter = HCOM_COMM_PROTOCOL_STR_MAP.find(protocol);
89 0 : if (iter == HCOM_COMM_PROTOCOL_STR_MAP.end()) {
90 0 : return "CommProtocol(" + std::to_string(protocol) + ")";
91 : } else {
92 0 : return iter->second;
93 : }
94 : }
95 :
96 : } // namespace MyRankUtils
97 :
98 : namespace hccl {
99 :
100 : constexpr uint32_t UNREUSE_CHANNEL_IDX = 0xFFFFFFFF;
101 :
102 167 : MyRank::MyRank(aclrtBinHandle binHandle, uint32_t rankId, const CommConfig &config, const ManagerCallbacks &callbacks,
103 : RankGraph *rankGraph,
104 167 : const Hccl::RankIpPortMapPtr& rankIpPortMap)
105 167 : : binHandle_(binHandle),
106 167 : rankId_(rankId),
107 167 : config_(config),
108 167 : callbacks_(callbacks),
109 167 : rankGraph_(rankGraph),
110 167 : rankIpPortMap_(rankIpPortMap)
111 : {
112 167 : }
113 :
114 334 : MyRank::~MyRank()
115 : {
116 167 : HCCL_INFO("[MyRank][~MyRank] MyRank deinit, rankId_[%u], devLogicId_[%d]", rankId_, devLogicId_);
117 : // 析构有时序要求
118 167 : rankPairMgr_ = nullptr; // 内部会销毁channel，可能需要返还endpoint与ccu资源
119 167 : endpointMgr_ = nullptr; // 内部会销毁endpoint，可能需要返回ccu资源
120 :
121 : struct ResourceCleanupGuard {
122 167 : explicit ResourceCleanupGuard(MyRank &myRank) : myRank_(myRank) {}
123 167 : ~ResourceCleanupGuard() noexcept
124 : {
125 167 : myRank_.ccuInsHandle_ = 0;
126 :
127 167 : if (!myRank_.useCcuResStaticAlloc_ && myRank_.ccuDrvHandle_) {
128 1 : myRank_.ccuDrvHandle_ = nullptr; // 先减少引用计数，再尝试关闭
129 1 : (void)CcuDeinitFeature(myRank_.devLogicId_);
130 : // 尝试关闭CCU功能，最后一个调用时会关闭CCU驱动
131 : }
132 :
133 167 : myRank_.ReleaseCcuMsCommReservation();
134 :
135 167 : myRank_.commMems_ = nullptr;
136 167 : myRank_.nsRecoveryProcessor_ = nullptr;
137 167 : }
138 :
139 : MyRank &myRank_;
140 167 : } cleanupGuard(*this);
141 :
142 167 : if (ccuInsHandle_ != 0) { // 内部清理CCU资源，关闭CCU通道
143 : // 刷新并获取当前线程的 DeviceId
144 6 : int32_t threadDevId = INVALID_INT;
145 6 : CHK_RET_NULL(HcclDeviceRefresh(threadDevId));
146 6 : HCCL_INFO("[%s] curDeviceLogicId[%d], threadDevId[%d]", __func__, devLogicId_, threadDevId);
147 : // 先切换为目标 curDeviceLogicId
148 6 : bool isDiffDevId = false;
149 6 : if (devLogicId_ != threadDevId) {
150 0 : CHK_RET_NULL(hrtSetDevice(devLogicId_));
151 0 : isDiffDevId = true;
152 : }
153 : // 销毁 CcuInstance
154 6 : CcuResult ret = HcommCcuInsDestroy(ccuInsHandle_);
155 6 : if (ret != CCU_SUCCESS) {
156 4 : HCCL_ERROR("[%s] HcommCcuInsDestroy failed, ret[%d]", __func__, ret);
157 : }
158 : // 切换回原来的 DeviceId
159 6 : if (isDiffDevId) {
160 0 : CHK_RET_NULL(hrtSetDevice(threadDevId));
161 0 : CHK_PRT(HcclDeviceRefresh(threadDevId));
162 : }
163 : }
164 334 : }
165 :
166 5 : HcclResult MyRank::GetLocalTlsStatus(Hccl::TlsStatus &tlsStatus) const
167 : {
168 5 : tlsStatus = Hccl::TlsStatus::UNKNOWN;
169 5 : s32 deviceLogicId = -1;
170 5 : u32 devicePhyId = INVALID_UINT;
171 5 : CHK_RET(hrtGetDevice(&deviceLogicId));
172 4 : CHK_RET(hrtGetDevicePhyIdByIndex(static_cast<u32>(deviceLogicId), devicePhyId));
173 :
174 3 : RaInfo info{};
175 3 : info.mode = NetworkMode::NETWORK_OFFLINE;
176 3 : info.phyId = devicePhyId;
177 3 : return Hccl::HrtRaGetTlsStatus(&info, tlsStatus);
178 : }
179 :
180 : constexpr uint32_t DEFAULT_MODE = 0;
181 : constexpr uint32_t AICPU_TS_MODE = 2;
182 : constexpr uint32_t CCU_MS_MODE = 5;
183 : constexpr uint32_t CCU_SCHED_MODE = 6;
184 13 : inline CcuInstanceType OpExpansionModeToCcuInstanceType(uint32_t opExpansionMode)
185 : {
186 : // 仅作数据类型转换，不做逻辑处理
187 13 : if (opExpansionMode == CCU_SCHED_MODE) {
188 4 : return CcuInstanceType::CCU_SCHED;
189 : }
190 :
191 9 : if (opExpansionMode == CCU_MS_MODE) {
192 7 : return CcuInstanceType::CCU_MS;
193 : }
194 :
195 2 : return CcuInstanceType::CCU_UNUSED;
196 : }
197 :
198 11 : HcclResult MyRank::TryInitCcuInstanceLegacy()
199 : {
200 11 : auto ccuInsType = OpExpansionModeToCcuInstanceType(opExpansionMode_);
201 11 : if (ccuInsType == CcuInstanceType::CCU_UNUSED) {
202 0 : ccuInsHandle_ = 0;
203 0 : return HcclResult::HCCL_SUCCESS;
204 : }
205 :
206 11 : auto ccuInitRet = HcommCcuInsCreateLegacy(ccuInsType, &ccuInsHandle_);
207 :
208 : // ccu驱动拉起失败，直接回退至aicpu ts
209 11 : if (ccuInitRet == CcuResult::CCU_E_DRV_BUSY) {
210 2 : opExpansionMode_ = AICPU_TS_MODE;
211 2 : ccuInsHandle_ = 0;
212 2 : HCCL_RUN_WARNING("[MyRank][%s] failed to init ccu driver, fallback to aicpu, rankId[%u].", __func__, rankId_);
213 2 : return HcclResult::HCCL_SUCCESS;
214 : }
215 :
216 : // ccu通信域数量过多，导致资源不足
217 9 : if (CCU_CHK_RES_UNAVAIL(ccuInitRet)) {
218 : // 如果是ccu ms模式，回退至ccu调度模式重试
219 : // 复用原有的ccuResContainer，回退到ccu sched时不需要重复拉起ccu驱动
220 4 : if (opExpansionMode_ == CCU_MS_MODE) {
221 3 : opExpansionMode_ = CCU_SCHED_MODE;
222 3 : CHK_RET(TryInitCcuInstanceLegacy()); // 至多递归一次
223 3 : return HcclResult::HCCL_SUCCESS;
224 : }
225 :
226 : // 其余模式资源不足回退至aicpu ts
227 1 : opExpansionMode_ = AICPU_TS_MODE;
228 1 : ccuInsHandle_ = 0;
229 1 : HCCL_RUN_WARNING("[MyRank][%s] ccu resources are unavailable, fallback to aicpu, rankId[%u].", __func__, rankId_);
230 1 : return HcclResult::HCCL_SUCCESS;
231 : }
232 :
233 : // 预期外返回值属于错误
234 5 : if (ccuInitRet != CcuResult::CCU_SUCCESS) {
235 2 : HCCL_ERROR("[%s] failed, ret[%d] is not expected.", __func__, ccuInitRet);
236 2 : ccuInsHandle_ = 0;
237 2 : return static_cast<HcclResult>(ccuInitRet);
238 : }
239 :
240 : // ccu资源申请成功
241 3 : return HcclResult::HCCL_SUCCESS;
242 : }
243 :
244 10 : HcclResult MyRank::ReserveCcuMsCommOrFallback()
245 : {
246 10 : if (opExpansionMode_ != CCU_MS_MODE) {
247 2 : return HCCL_SUCCESS;
248 : }
249 :
250 8 : bool reserved = false;
251 8 : CHK_RET(CollCommMgr::GetInstance()->TryReserveCcuMsComm(
252 : devLogicId_, config_.GetConfigCommName(), reserved));
253 8 : if (reserved) {
254 7 : ccuMsCommReserved_ = true;
255 7 : return HCCL_SUCCESS;
256 : }
257 :
258 1 : opExpansionMode_ = CCU_SCHED_MODE;
259 1 : HCCL_RUN_WARNING("[MyRank][%s] CCU_MS comm already exists on device[%d], fallback to CCU_SCHED, rankId[%u].",
260 : __func__, devLogicId_, rankId_);
261 1 : return HCCL_SUCCESS;
262 : }
263 :
264 177 : void MyRank::ReleaseCcuMsCommReservation()
265 : {
266 177 : if (!ccuMsCommReserved_) {
267 169 : return;
268 : }
269 8 : CollCommMgr::GetInstance()->ReleaseCcuMsComm(devLogicId_, config_.GetConfigCommName());
270 8 : ccuMsCommReserved_ = false;
271 : }
272 :
273 10 : void MyRank::ReconcileCcuMsCommReservation(HcclResult initRet)
274 : {
275 10 : if (initRet != HCCL_SUCCESS || opExpansionMode_ != CCU_MS_MODE) {
276 10 : ReleaseCcuMsCommReservation();
277 : }
278 10 : }
279 :
280 2 : HcclResult MyRank::TryInitCcuInstanceOnDemand()
281 : {
282 : // 以下为ccu新接口流程
283 2 : auto ccuInsType = OpExpansionModeToCcuInstanceType(opExpansionMode_);
284 2 : if (ccuInsType == CcuInstanceType::CCU_UNUSED) {
285 2 : ccuInsHandle_ = 0;
286 2 : return HcclResult::HCCL_SUCCESS;
287 : }
288 :
289 0 : if (mainBoardType_ == Hccl::HcclMainboardId::MAINBOARD_OTHERS) {
290 0 : CHK_RET(CcuGetMainboardType(devLogicId_, mainBoardType_));
291 : }
292 :
293 0 : if (mainBoardType_ == Hccl::HcclMainboardId::MAINBOARD_PCIE_STD &&
294 : ccuInsType == CcuInstanceType::CCU_MS) { // 标卡环境下配置CCU_MS拦截报错
295 0 : HCCL_ERROR("[%s] ccuInstanceType[%d] not support in %s", __func__,
296 : ccuInsType, mainBoardType_.Describe().c_str());
297 0 : return HcclResult::HCCL_E_NOT_SUPPORT;
298 : }
299 :
300 : // 拉起ccu驱动
301 0 : if (!ccuDrvHandle_) {
302 0 : auto ccuInitRet = CcuInitFeature(devLogicId_, ccuDrvHandle_);
303 :
304 : // ccu驱动拉起失败，直接回退至aicpu ts
305 0 : if (ccuInitRet == CcuResult::CCU_E_DRV_BUSY) {
306 0 : opExpansionMode_ = AICPU_TS_MODE;
307 0 : ccuInsHandle_ = 0;
308 0 : HCCL_RUN_WARNING("[MyRank][%s] failed to init ccu driver, "
309 : "fallback to aicpu, rankId[%u].", __func__, rankId_);
310 0 : return HcclResult::HCCL_SUCCESS;
311 : }
312 :
313 : // 预期外返回值属于错误
314 0 : if (ccuInitRet != CcuResult::CCU_SUCCESS) {
315 0 : HCCL_ERROR("[%s] failed, ret[%d] is not expected.",
316 : __func__, ccuInitRet);
317 0 : ccuInsHandle_ = 0;
318 0 : return static_cast<HcclResult>(ccuInitRet);
319 : }
320 : }
321 :
322 : // ccu驱动拉起成功
323 0 : return HcclResult::HCCL_SUCCESS;
324 : }
325 :
326 10 : HcclResult MyRank::TryInitCcuInstance()
327 : {
328 10 : CHK_RET(ReserveCcuMsCommOrFallback());
329 :
330 10 : HcclResult ret = HCCL_SUCCESS;
331 10 : if (useCcuResStaticAlloc_) {
332 8 : HCCL_RUN_INFO(
333 : "[MyRank][%s] HCCL version does not support CCU on-demand resource allocation, use legacy allocation.",
334 : __func__);
335 8 : ret = TryInitCcuInstanceLegacy();
336 : } else {
337 2 : ret = TryInitCcuInstanceOnDemand();
338 : }
339 10 : ReconcileCcuMsCommReservation(ret);
340 10 : return ret;
341 : }
342 :
343 33 : HcclResult MyRank::GetDevicePortInternal(uint32_t rank, uint32_t *devPort, EndpointLocType locType)
344 : {
345 33 : CHK_PTR_NULL(devPort);
346 33 : CHK_PTR_NULL(rankGraph_);
347 :
348 : DevType devType;
349 33 : CHK_RET(hrtGetDeviceType(devType));
350 : // v1 模式 (mode_ == 0): 强制转换为 RankGraphV1 调用 GetDevicePort
351 : // v2 模式 (mode_ != 0): 使用 rankGraph_->GetDevicePort()
352 33 : if (devType == DevType::DEV_TYPE_910B) {
353 0 : RankGraphV1* rankGraphV1 = static_cast<RankGraphV1*>(rankGraph_);
354 0 : CHK_RET(rankGraphV1->GetDevicePort(rank, devPort));
355 : } else {
356 33 : CHK_RET(rankGraph_->GetListenPort(rank, devPort, locType));
357 : }
358 33 : return HCCL_SUCCESS;
359 : }
360 :
361 110 : inline HcclResult GetHcclVersion(int &hcclVersion)
362 : {
363 110 : char hcclPkgName[] = "hccl";
364 110 : aclError aclRet = aclsysGetVersionNum(hcclPkgName, &hcclVersion);
365 110 : CHK_PRT_RET(aclRet != ACL_SUCCESS,
366 : HCCL_ERROR("[GetHcclVersion] aclsysGetVersionNum failed, aclRet[%d].", aclRet), HCCL_E_INTERNAL);
367 110 : HCCL_RUN_INFO("[GetHcclVersion] hccl version is %d.", hcclVersion);
368 110 : return HCCL_SUCCESS;
369 : }
370 :
371 : constexpr int MAX_HCCL_VERSION_USING_CCU_RES_STATIC_ALLOC = 90100000;
372 110 : HcclResult MyRank::Init(HcclMem cclBuffer, const uint32_t opExpansionMode, uint32_t rankNum)
373 : {
374 : // EXCEPTION_HANDLE_BEGIN
375 110 : CHK_RET(hrtGetDevice(&devLogicId_));
376 :
377 : // 获取hccl版本
378 110 : int hcclVersion = 0;
379 110 : CHK_RET(GetHcclVersion(hcclVersion));
380 110 : useCcuResStaticAlloc_ = hcclVersion <= MAX_HCCL_VERSION_USING_CCU_RES_STATIC_ALLOC;
381 :
382 : // ns recovery processor初始化
383 110 : EXCEPTION_CATCH(nsRecoveryProcessor_ = std::make_unique<NsRecoveryProcessor>(), return HCCL_E_PTR);
384 :
385 : // 创建通信内存管理器
386 110 : EXCEPTION_CATCH(commMems_ = std::make_unique<CommMems>(config_.GetConfigBufferSize()), return HCCL_E_PTR);
387 :
388 : // 初始化通信内存
389 110 : CHK_RET(commMems_->Init(cclBuffer));
390 :
391 110 : EXCEPTION_CATCH(engineCtxs_ = std::make_unique<EngineCtxs>(), return HCCL_E_PTR);
392 :
393 : // 通信域配置config优先级更高，当配置默认展开模式时，读取环境变量配置
394 110 : opExpansionMode_ = opExpansionMode;
395 110 : if (opExpansionMode_ == DEFAULT_MODE) {
396 : // 环境变量模块已处理，当用户未配置时，输出ccu sched模式
397 1 : auto accelerator = Hccl::EnvConfig::GetInstance().GetAlgoConfig().GetHcclAccelerator();
398 1 : HCCL_RUN_INFO("[MyRank][%s] set op expansion mode by env[%s].",
399 : __func__, accelerator.Describe().c_str());
400 1 : opExpansionMode_ = static_cast<uint32_t>(accelerator);
401 : }
402 :
403 : // 仅自定义算子ccu流程初始化资源
404 110 : if (ccuInsHandle_ == 0 && rankNum != 1 &&
405 109 : (opExpansionMode_ == CCU_MS_MODE || opExpansionMode_ == CCU_SCHED_MODE)) {
406 7 : const uint32_t originOpExpansionMode = opExpansionMode_; // 记录原始加速模式，避免中间执行修改后丢失
407 7 : auto ret = TryInitCcuInstance();
408 7 : if (ret != HcclResult::HCCL_SUCCESS) { // 申请成功与回退成功都属于成功，其他均非预期
409 1 : HCCL_ERROR("[MyRank][%s] failed to init ccu instance, op expansion mode[%u].",
410 : __func__, originOpExpansionMode);
411 1 : return ret;
412 : }
413 : }
414 :
415 : // 创建端点管理器
416 109 : EXCEPTION_CATCH(endpointMgr_ = std::make_unique<hcomm::EndpointMgr>(), return HCCL_E_PTR);
417 :
418 : // rankPairMgr_初始化
419 109 : EXCEPTION_CATCH(rankPairMgr_ = std::make_unique<RankPairMgr>(rankIpPortMap_), return HCCL_E_PTR);
420 :
421 109 : DlProfFunction::GetInstance().DlProfFunctionInit();
422 : // EXCEPTION_HANDLE_END
423 109 : return HCCL_SUCCESS;
424 : }
425 :
426 9 : HcclResult MyRank::QueryListenPort(uint32_t localRank, uint32_t remoteRank, const EndpointDesc &localEndpointDesc,
427 : const EndpointDesc &remoteEndpointDesc, uint32_t &listenPort, HcommChannelDesc &hcommDesc)
428 : {
429 : // 查询rmtRankId对应的devPort
430 9 : uint32_t rmtPort = 0;
431 9 : CHK_RET(GetDevicePortInternal(remoteRank, &rmtPort, remoteEndpointDesc.loc.locType));
432 9 : if (rmtPort > Hccl::MAX_VALUE_TCPPORT) {
433 1 : HCCL_ERROR("[%s] Invalid port[%u] of Rank[%u]", __func__, rmtPort, remoteRank);
434 1 : return HCCL_E_PARA;
435 : }
436 : // 查询该socket链接的server端监听的端口（监听方的选择策略需要跟SocketConfig中保持一致）
437 8 : Hccl::IpAddress localIpAddr{};
438 8 : Hccl::IpAddress remoteIpAddr{};
439 8 : CHK_RET(CommAddrToIpAddress(localEndpointDesc.commAddr, localIpAddr));
440 8 : CHK_RET(CommAddrToIpAddress(remoteEndpointDesc.commAddr, remoteIpAddr));
441 8 : if (localIpAddr < remoteIpAddr) {
442 : // 查询localRankId对应的devPort
443 6 : CHK_RET(GetDevicePortInternal(localRank, &listenPort, localEndpointDesc.loc.locType));
444 6 : hcommDesc.role = HcommSocketRole::HCOMM_SOCKET_ROLE_SERVER;
445 6 : if (listenPort > Hccl::MAX_VALUE_TCPPORT) {
446 0 : HCCL_ERROR("[%s] Invalid port[%u] of Rank[%u]", __func__, listenPort, localRank);
447 0 : return HCCL_E_PARA;
448 : }
449 6 : hcommDesc.port = static_cast<uint16_t>(listenPort); // HcommChannelDesc.port中填监听端口号
450 : } else {
451 2 : listenPort = rmtPort;
452 2 : hcommDesc.role = HcommSocketRole::HCOMM_SOCKET_ROLE_CLIENT;
453 2 : hcommDesc.port = static_cast<uint16_t>(rmtPort); // HcommChannelDesc.port中填对端端口号(此场景下对端端口号也就是监听端口号)
454 : }
455 :
456 8 : return HCCL_SUCCESS;
457 : }
458 :
459 12 : HcclResult MyRank::GetEndpointPairFromChannel(const HcclChannelDesc &channelDesc, uint32_t channelIndex, uint32_t channelNum,
460 : uint32_t &remoteRank, hcomm::EndpointPair* &endpointPair, RankPair* &rankPair)
461 : {
462 12 : remoteRank = channelDesc.remoteRank;
463 12 : HCCL_INFO("[%s][%u/%u] remoteRank[%u] localProtocol[%d] remoteProtocol[%d]",
464 : __func__, channelIndex + 1, channelNum, remoteRank, channelDesc.localEndpoint.protocol, channelDesc.remoteEndpoint.protocol);
465 :
466 12 : const RankIdPair rankIdPair = std::make_pair(rankId_, remoteRank);
467 12 : const EndpointDescPair endpointDescPair = std::make_pair(channelDesc.localEndpoint, channelDesc.remoteEndpoint);
468 12 : CHK_RET(rankPairMgr_->Get(rankIdPair, rankPair));
469 12 : CHK_PTR_NULL(rankPair);
470 12 : CHK_RET(rankPair->GetEndpointPair(endpointDescPair, endpointPair));
471 12 : CHK_PTR_NULL(endpointPair);
472 12 : return HCCL_SUCCESS;
473 : }
474 :
475 12 : inline std::string AddProtocolToSocketTag(const std::string &socketTag, const HcclChannelDesc* channelDescs)
476 : {
477 12 : std::string newSocketTag = socketTag + "_protocol_" + std::to_string(channelDescs->channelProtocol);
478 12 : return newSocketTag;
479 : }
480 :
481 3 : HcclResult MyRank::BatchServerInitForChannels(const HcclChannelDesc* channelDescs, uint32_t channelNum,
482 : const std::string &socketTag, ReuseSocketIdxMap &reuseSocketIdxMap)
483 : {
484 : // 批量获取socket，与server监听隔离开
485 9 : for (uint32_t i = 0; i < channelNum; ++i) {
486 6 : hcomm::EndpointPair* endpointPair = nullptr;
487 6 : RankPair* rankPair = nullptr;
488 6 : uint32_t remoteRank = 0;
489 :
490 6 : CHK_RET(GetEndpointPairFromChannel(channelDescs[i], i, channelNum, remoteRank, endpointPair, rankPair));
491 :
492 6 : if (reuseSocketIdxMap.find(rankPair) == reuseSocketIdxMap.end()) {
493 4 : std::unordered_map<hcomm::EndpointPair*, u32> endpointPair2Idx{};
494 4 : endpointPair2Idx.emplace(endpointPair, 0);
495 4 : reuseSocketIdxMap.emplace(rankPair, endpointPair2Idx);
496 6 : } else if (reuseSocketIdxMap[rankPair].find(endpointPair) == reuseSocketIdxMap[rankPair].end()) {
497 0 : reuseSocketIdxMap[rankPair].emplace(endpointPair, 0);
498 : }
499 6 : u32& reuseIdx = reuseSocketIdxMap[rankPair][endpointPair];
500 :
501 : uint32_t devicePhyId;
502 : uint32_t remoteDevicePhyId;
503 6 : rankGraph_->GetDeviceId(rankId_, &devicePhyId);
504 6 : rankGraph_->GetDeviceId(remoteRank, &remoteDevicePhyId);
505 :
506 6 : const std::string socketTagAddProto = AddProtocolToSocketTag(socketTag, &channelDescs[i]);
507 6 : auto ret = endpointPair->ServerInit(rankId_, remoteRank, socketTagAddProto, reuseIdx, devicePhyId, remoteDevicePhyId);
508 6 : CHK_PRT_RET(ret != HCCL_SUCCESS,
509 : HCCL_ERROR("[%s] ServerInitFailed, channelIndex[%u], remoteRank[%u], protocol[%d] reuseIdx[%u]",
510 : __func__, i, remoteRank, channelDescs[i].localEndpoint.protocol, reuseIdx),
511 : ret);
512 :
513 6 : HCCL_INFO("[%s][%u/%u] server listen successfully, remoteRank[%u], reuseIdx[%u]",
514 : __func__, i + 1, channelNum, remoteRank, reuseIdx);
515 6 : }
516 3 : return HCCL_SUCCESS;
517 : }
518 :
519 3 : HcclResult MyRank::BatchGetSocketsForChannels(const HcclChannelDesc* channelDescs, uint32_t channelNum,
520 : const std::string &socketTag, std::vector<HcommChannelDesc> &hcommDescs,
521 : ReuseSocketIdxMap &reuseSocketIdxMap)
522 : {
523 9 : for (uint32_t i = 0; i < channelNum; ++i) {
524 6 : hcomm::EndpointPair* endpointPair = nullptr;
525 6 : RankPair* rankPair = nullptr;
526 6 : uint32_t remoteRank = 0;
527 :
528 6 : CHK_RET(GetEndpointPairFromChannel(channelDescs[i], i, channelNum, remoteRank, endpointPair, rankPair));
529 :
530 6 : uint32_t listenPort = 0;
531 6 : CHK_RET(QueryListenPort(rankId_, remoteRank, channelDescs[i].localEndpoint, channelDescs[i].remoteEndpoint, listenPort, hcommDescs[i]));
532 :
533 6 : u32& reuseIdx = reuseSocketIdxMap[rankPair][endpointPair];
534 : uint32_t devicePhyId;
535 : uint32_t remoteDevicePhyId;
536 6 : rankGraph_->GetDeviceId(rankId_, &devicePhyId);
537 6 : rankGraph_->GetDeviceId(remoteRank, &remoteDevicePhyId);
538 6 : HCCL_INFO("[MyRank][BatchCreateSockets] rankId_[%u] devicePhyId[%u] remoteRank[%u] remoteDevicePhyId[%u]",
539 : rankId_, devicePhyId, remoteRank, remoteDevicePhyId);
540 6 : Hccl::Socket* socket = nullptr;
541 6 : const std::string socketTagAddProto = AddProtocolToSocketTag(socketTag, &channelDescs[i]);
542 6 : auto ret = endpointPair->GetConnectedSocket(rankId_, remoteRank, socketTagAddProto, reuseIdx, listenPort, socket, devicePhyId, remoteDevicePhyId);
543 6 : CHK_PRT_RET(ret != HCCL_SUCCESS,
544 : HCCL_ERROR("[%s] failed to get socket, channelIndex[%u], remoteRank[%u], protocol[%d], reuseIdx[%u], tag[%s]",
545 : __func__, i, remoteRank, channelDescs[i].localEndpoint.protocol, reuseIdx, socketTagAddProto.c_str()),
546 : ret);
547 6 : CHK_PTR_NULL(socket);
548 :
549 6 : hcommDescs[i].socket = reinterpret_cast<HcommSocket>(socket);
550 :
551 6 : HCCL_INFO("[%s][%u/%u] socket created successfully, remoteRank[%u], socket[%p] reuseIdx[%u]",
552 : __func__, i + 1, channelNum, remoteRank, socket, reuseIdx);
553 6 : reuseIdx++;
554 6 : }
555 3 : return HCCL_SUCCESS;
556 : }
557 :
558 3 : HcclResult MyRank::BatchCreateSockets(const HcclChannelDesc* channelDescs, uint32_t channelNum,
559 : const std::string &socketTag, std::vector<HcommChannelDesc> &hcommDescs)
560 : {
561 3 : CHK_PTR_NULL(channelDescs);
562 3 : CHK_PRT_RET(channelNum == 0,
563 : HCCL_ERROR("[%s] invalid param: channelNum is zero", __func__), HCCL_E_PARA);
564 :
565 3 : ReuseSocketIdxMap reuseSocketIdxMap{};
566 : // socket服务器首先监听
567 3 : CHK_RET(BatchServerInitForChannels(channelDescs, channelNum, socketTag, reuseSocketIdxMap));
568 : // socket添加白名单以及进行连接，获取最后的socket
569 3 : CHK_RET(BatchGetSocketsForChannels(channelDescs, channelNum, socketTag, hcommDescs, reuseSocketIdxMap));
570 3 : return HCCL_SUCCESS;
571 3 : }
572 :
573 : constexpr uint32_t MEM_HANDLE_NUM_MAX = 256; // memHandleNum的默认限制最大为256
574 : constexpr uint32_t NOTIFY_NUM_MAX = 64; // notifynum 的默认限制最大为64
575 :
576 2 : HcclResult MyRank::CheckChannelParam(CommEngine engine, const HcclChannelDesc* channelDesc,
577 : uint32_t channelNum) const
578 : {
579 3 : for (u32 index = 0; index < channelNum; ++index) {
580 2 : if (engine == COMM_ENGINE_AIV) {
581 0 : CHK_PRT_RET(
582 : (channelDesc->memHandleNum > MEM_HANDLE_NUM_MAX),
583 : HCCL_ERROR("[%s]Channeldesc[%u] invalid memHandleNum, memHandleNum[%u], max channel num[%u]",
584 : __func__, index, channelDesc->memHandleNum, MEM_HANDLE_NUM_MAX), HCCL_E_PARA
585 : );
586 0 : CHK_PRT_RET(
587 : (channelDesc->memHandleNum != 0 && channelDesc->memHandles == nullptr),
588 : HCCL_ERROR("[%s]Channeldesc[%u] invalid memHandles, memHandles is null",
589 : __func__, index), HCCL_E_PARA
590 : );
591 : } else {
592 2 : if (channelDesc->memHandleNum != 0) {
593 1 : HCCL_WARNING("[%s]Channeldesc[%u] memHandleNum[%u] is non-zero, memHandle exchange is not supported.",
594 : __func__, index, channelDesc->memHandleNum);
595 : }
596 : }
597 2 : CHK_PRT_RET(channelDesc->notifyNum > NOTIFY_NUM_MAX,
598 : HCCL_ERROR("[%s]Channeldesc[%u] invalid notifyNum [%u], max notify num[%u]",
599 : __func__, index, channelDesc->notifyNum, NOTIFY_NUM_MAX), HCCL_E_PARA);
600 : }
601 :
602 1 : return HCCL_SUCCESS;
603 : }
604 :
605 : // 批量创建channels，如果CCU资源不足（如Xn, Cke, channel ctx, jetty ctx, wqebb）会失败，返回HCCL_E_UNAVAIL
606 6 : HcclResult MyRank::BatchCreateChannels(CommEngine engine, const HcclChannelDesc* channelDescs, uint32_t channelNum,
607 : std::vector<HcommChannelDesc> &hcommDescs, ChannelHandle *channelHandles)
608 : {
609 6 : CHK_PTR_NULL(channelDescs);
610 6 : CHK_PTR_NULL(channelHandles);
611 6 : CHK_PRT_RET(channelNum == 0,
612 : HCCL_ERROR("[%s] invalid param: channelNum is zero", __func__), HCCL_E_PARA);
613 :
614 6 : uint32_t localRank = rankId_;
615 6 : std::vector<HcclMem> memVec;
616 6 : CHK_SMART_PTR_NULL(commMems_);
617 6 : CHK_RET(commMems_->GetMemoryHandles(memVec));
618 : std::unordered_map<RankPair*, std::unordered_map<CommEngine,
619 6 : std::unordered_map<hcomm::EndpointPair*, u32>>> reuseChannelIdxMap{};
620 :
621 : // 记录本轮新申请的channel
622 6 : newChannels_.clear();
623 6 : bool isAllSuccess = true;
624 :
625 22 : for (uint32_t i = 0; i < channelNum; ++i) {
626 18 : const EndpointDesc &localEndpointDesc = channelDescs[i].localEndpoint;
627 18 : const EndpointDesc &remoteEndpointDesc = channelDescs[i].remoteEndpoint;
628 18 : uint32_t remoteRank = channelDescs[i].remoteRank;
629 :
630 18 : HCCL_INFO("[%s][%u/%u] remoteRank[%u] localProtocol[%d] remoteProtocol[%d] engine[%s]",
631 : __func__, i + 1, channelNum, remoteRank, localEndpointDesc.protocol, remoteEndpointDesc.protocol, GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str()
632 : );
633 :
634 18 : EndpointHandle epHandle = nullptr;
635 18 : CHK_PTR_NULL(endpointMgr_);
636 18 : auto ret = endpointMgr_->Get(localEndpointDesc, epHandle);
637 18 : CHK_PRT_RET(ret != HCCL_SUCCESS,
638 : HCCL_ERROR("[%s] failed to get endpoint, channelIndex[%u], remoteRank[%u], protocol[%d]",
639 : __func__, i, remoteRank, localEndpointDesc.protocol),
640 : ret);
641 18 : CHK_PTR_NULL(epHandle);
642 :
643 : // 启动监听
644 18 : uint32_t listenPort = 0;
645 18 : CHK_RET(GetDevicePortInternal(localRank, &listenPort, localEndpointDesc.loc.locType));
646 18 : CHK_RET(static_cast<HcclResult>(HcommEndpointStartListen(epHandle, listenPort, nullptr)));
647 :
648 18 : HCCL_INFO("[%s][%u/%u] remoteRank[%u] epHandle[%p] protocol[%d]",
649 : __func__, i + 1, channelNum, remoteRank,
650 : epHandle, localEndpointDesc.protocol);
651 :
652 : // 注册内存
653 18 : std::vector<MemHandle> memHandleVec;
654 18 : std::vector<std::string> memTag;
655 18 : memVec.clear();
656 18 : CHK_RET(commMems_->GetTagMemoryHandles(channelDescs[i].memHandles, channelDescs[i].memHandleNum, memVec, memTag));
657 18 : HCCL_INFO("[%s][%u/%u] remoteRank[%u] got %zu user memory handles",
658 : __func__, i + 1, channelNum, remoteRank, memVec.size());
659 18 : ret = endpointMgr_->RegisterMemory(epHandle, memTag, memVec, memHandleVec);
660 18 : CHK_PRT_RET(ret != HCCL_SUCCESS,
661 : HCCL_ERROR("[%s] failed to register memory, channelIndex[%u], remoteRank[%u], memTagNum[%zu]",
662 : __func__, i, remoteRank, memTag.size()),
663 : ret);
664 :
665 18 : hcommDescs[i].exchangeAllMems = false;
666 18 : hcommDescs[i].memHandles = memHandleVec.data();
667 18 : hcommDescs[i].memHandleNum = memHandleVec.size();
668 :
669 18 : hcomm::EndpointPair* endpointPair = nullptr;
670 18 : RankIdPair rankIdPair = std::make_pair(localRank, remoteRank);
671 18 : EndpointDescPair endpointDescPair = std::make_pair(localEndpointDesc, remoteEndpointDesc);
672 18 : RankPair* rankPair = nullptr;
673 18 : CHK_RET(rankPairMgr_->Get(rankIdPair, rankPair));
674 18 : CHK_PTR_NULL(rankPair);
675 18 : CHK_RET(rankPair->GetEndpointPair(endpointDescPair, endpointPair));
676 18 : CHK_PTR_NULL(endpointPair);
677 :
678 18 : if (reuseChannelIdxMap.find(rankPair) == reuseChannelIdxMap.end()) {
679 10 : std::unordered_map<CommEngine, std::unordered_map<hcomm::EndpointPair*, u32>> engine2EndpointPairMap{};
680 10 : std::unordered_map<hcomm::EndpointPair*, u32> endpointPair2Idx{};
681 10 : endpointPair2Idx.emplace(endpointPair, 0);
682 10 : engine2EndpointPairMap.emplace(engine, endpointPair2Idx);
683 10 : reuseChannelIdxMap.emplace(rankPair, engine2EndpointPairMap);
684 18 : } else if (reuseChannelIdxMap[rankPair].find(engine) == reuseChannelIdxMap[rankPair].end()) {
685 0 : std::unordered_map<hcomm::EndpointPair*, u32> endpointPair2Idx{};
686 0 : endpointPair2Idx.emplace(endpointPair, 0);
687 0 : reuseChannelIdxMap[rankPair].emplace(engine, endpointPair2Idx);
688 8 : } else if (reuseChannelIdxMap[rankPair][engine].find(endpointPair) == reuseChannelIdxMap[rankPair][engine].end()) {
689 0 : reuseChannelIdxMap[rankPair][engine].emplace(endpointPair, 0);
690 : }
691 :
692 18 : u32& reuseIdx = reuseChannelIdxMap[rankPair][engine][endpointPair];
693 18 : u32 idx = reuseIdx;
694 : /* hostNIC -- DeviceNic（transport不复用link/Channel），此流程也是新创建channel，需要计入isNewChannel */
695 18 : if (localEndpointDesc.loc.locType != remoteEndpointDesc.loc.locType) {
696 0 : idx = UNREUSE_CHANNEL_IDX;
697 : }
698 18 : bool isNewChannel = (endpointPair->IsChannelNotExist(engine, reuseIdx) || (idx == UNREUSE_CHANNEL_IDX));
699 :
700 : // CreateChannel 返回 HCCL_E_UNAVAIL 表示资源不足创建失败
701 18 : ret = endpointPair->CreateChannel(epHandle, engine, idx, &hcommDescs[i], channelHandles + i);
702 18 : if (ret == HCCL_E_TIMEOUT || ret == HCCL_E_INTERNAL) {
703 0 : Hccl::TlsStatus tlsStatus = Hccl::TlsStatus::UNKNOWN;
704 0 : CHK_PRT_CONT(GetLocalTlsStatus(tlsStatus) != HCCL_SUCCESS,
705 : HCCL_WARNING("[GetLocalTlsStatus] Can not get TlsStatus"));
706 : }
707 18 : if (ret == HCCL_E_UNAVAIL) {
708 : // 申请channel因资源不足失败，清理已申请的channel
709 2 : HCCL_RUN_WARNING("[%s] create channel failed, channelIndex[%u], remoteRank[%u], engine[%s], reuseIdx[%u], need clean new channels",
710 : __func__, i + 1, remoteRank, GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), reuseIdx);
711 2 : isAllSuccess = false;
712 2 : break;
713 : }
714 : // 记录新申请的channel信息，用于清理临时资源
715 16 : if (isNewChannel) {
716 10 : newChannels_.emplace_back(std::make_pair(i, reuseIdx));
717 : }
718 :
719 16 : CHK_PRT_RET(ret != HCCL_SUCCESS,
720 : HCCL_ERROR("[%s] failed to create channel, channelIndex[%u], remoteRank[%u], engine[%s], reuseIndex[%u]",
721 : __func__, i + 1, remoteRank, GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), reuseIdx),
722 : ret);
723 16 : if (idx != UNREUSE_CHANNEL_IDX) {
724 16 : reuseIdx++;
725 : }
726 :
727 16 : HCCL_INFO("[%s][%u/%u] channel created successfully, remoteRank[%u], channelHandle[%p]",
728 : __func__, i + 1, channelNum, remoteRank, channelHandles[i]);
729 20 : }
730 :
731 : // 如果申请失败，清理endpoint pair中记录的channel handle
732 6 : if (!isAllSuccess) {
733 2 : HCCL_RUN_WARNING("[%s] create channel failed, destroy new channels num[%zu], engine[%s]", __func__, newChannels_.size(), GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str());
734 2 : CHK_RET(DestroyNewChannels(engine, channelDescs));
735 2 : return HCCL_E_UNAVAIL;
736 : }
737 :
738 4 : return HCCL_SUCCESS;
739 6 : }
740 :
741 2 : HcclResult MyRank::DestroyNewChannels(CommEngine engine, const HcclChannelDesc* channelDescs)
742 : {
743 2 : uint32_t localRank = rankId_;
744 14 : for (auto idxPairIter = std::rbegin(newChannels_); idxPairIter != std::rend(newChannels_); ++idxPairIter) { // 由于新申请的在申请过的后面，所以要从后往前找reuseIdx销毁
745 4 : auto idxPair = *idxPairIter;
746 4 : const EndpointDesc &localEndpointDesc = channelDescs[idxPair.first].localEndpoint;
747 4 : const EndpointDesc &remoteEndpointDesc = channelDescs[idxPair.first].remoteEndpoint;
748 4 : uint32_t remoteRank = channelDescs[idxPair.first].remoteRank;
749 4 : hcomm::EndpointPair* endpointPair = nullptr;
750 4 : RankIdPair rankIdPair = std::make_pair(localRank, remoteRank);
751 4 : EndpointDescPair endpointDescPair = std::make_pair(localEndpointDesc, remoteEndpointDesc);
752 4 : RankPair* rankPair = nullptr;
753 4 : CHK_RET(rankPairMgr_->Get(rankIdPair, rankPair));
754 4 : CHK_PTR_NULL(rankPair);
755 4 : CHK_RET(rankPair->GetEndpointPair(endpointDescPair, endpointPair));
756 4 : CHK_PTR_NULL(endpointPair);
757 4 : CHK_RET(endpointPair->DestroyChannel(engine, idxPair.second));
758 : }
759 2 : newChannels_.clear();
760 2 : return HCCL_SUCCESS;
761 : }
762 :
763 2 : HcclResult MyRank::BatchConnectChannels(const HcclChannelDesc* channelDescs, ChannelHandle *channelHandles, uint32_t channelNum)
764 : {
765 2 : auto timeout = std::chrono::seconds(Hccl::EnvConfig::GetInstance().GetSocketConfig().GetLinkTimeOut());
766 2 : auto startTime = std::chrono::steady_clock::now();
767 :
768 2 : HCCL_INFO("[%s] start connecting channels, channelNum[%u], timeout[%lld]sec",
769 : __func__, channelNum, timeout.count());
770 :
771 2 : std::vector<int32_t> statusVec(channelNum, 0);
772 2 : int32_t* statusList = statusVec.data();
773 2 : uint32_t retryCount = 0;
774 : while (true) {
775 1485356 : HcclResult ret = hcomm::ChannelProcess::ChannelGetStatus(channelHandles, channelNum, statusList);
776 :
777 : // 卫语句：先处理异常情况
778 :
779 : // 1. 检查超时
780 1485356 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
781 2 : auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
782 4 : std::chrono::steady_clock::now() - startTime).count();
783 2 : HCCL_ERROR("[%s] channel connect timeout after %lld sec, channelNum[%u], elapsed[%lld]ms, retryCount[%u]",
784 : __func__, timeout.count(), channelNum, elapsed, retryCount);
785 14 : RPT_INPUT_ERR(true, "EI0006", std::vector<std::string>({"reason"}), \
786 : std::vector<std::string>({GET_SOCKET_TIMEOUT_REASON_CLOSE_DETECT}));
787 2 : Hccl::TlsStatus tlsStatus = Hccl::TlsStatus::UNKNOWN;
788 2 : CHK_PRT_CONT(GetLocalTlsStatus(tlsStatus) != HCCL_SUCCESS,
789 : HCCL_WARNING("[GetLocalTlsStatus] Can not get TlsStatus"));
790 2 : logger::ChannelLogger::PrintChannelErrorDetails(
791 : rankId_, channelNum, channelDescs, channelHandles, statusList, static_cast<uint64_t>(elapsed), tlsStatus);
792 2 : return HCCL_E_TIMEOUT;
793 : }
794 :
795 : // 2. 处理重试（去除频繁的重试日志，一秒可能重试上千次）
796 1485354 : if (ret == HCCL_E_AGAIN) {
797 1485354 : retryCount++;
798 1485354 : continue;
799 : }
800 :
801 : // 3. 处理失败
802 0 : if (ret != HCCL_SUCCESS) {
803 0 : auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
804 0 : std::chrono::steady_clock::now() - startTime).count();
805 0 : HCCL_ERROR("[%s] channel connect failed, channelNum[%u], ret[%d], elapsed[%lld]ms, retryCount[%u]",
806 : __func__, channelNum, ret, elapsed, retryCount);
807 0 : Hccl::TlsStatus tlsStatus = Hccl::TlsStatus::UNKNOWN;
808 0 : CHK_PRT_CONT(GetLocalTlsStatus(tlsStatus) != HCCL_SUCCESS,
809 : HCCL_WARNING("[GetLocalTlsStatus] Can not get TlsStatus"));
810 0 : logger::ChannelLogger::PrintChannelErrorDetails(
811 : rankId_, channelNum, channelDescs, channelHandles, statusList, static_cast<uint64_t>(elapsed), tlsStatus);
812 0 : return ret;
813 : }
814 :
815 : // 4. 正常情况：所有通道连接成功
816 0 : auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
817 0 : std::chrono::steady_clock::now() - startTime).count();
818 0 : HCCL_INFO("[%s] all channels connected successfully, channelNum[%u], elapsed[%lld]ms, retryCount[%u]",
819 : __func__, channelNum, elapsed, retryCount);
820 0 : break;
821 1485354 : }
822 0 : return HCCL_SUCCESS;
823 4 : }
824 :
825 7 : HcclResult MyRank::ConfigSqDepthByExpansionMode(CommEngine engine, HcommChannelDesc& hcommDesc) const
826 : {
827 7 : constexpr u32 CCU_MS_MODE_DEPTH = 128;
828 7 : constexpr u32 CCU_SCHED_MODE_DEPTH = 16;
829 7 : if (engine == COMM_ENGINE_CCU) {
830 5 : if (opExpansionMode_ == CCU_MS_MODE) {
831 2 : hcommDesc.ubAttr.sqDepth = CCU_MS_MODE_DEPTH;
832 3 : } else if (opExpansionMode_ == CCU_SCHED_MODE) {
833 2 : hcommDesc.ubAttr.sqDepth = CCU_SCHED_MODE_DEPTH;
834 : } else {
835 1 : HCCL_ERROR("[%s] unexpected op expansion mode[%u] for ccu,", __func__, opExpansionMode_);
836 1 : return HCCL_E_INTERNAL;
837 : }
838 : }
839 6 : return HCCL_SUCCESS;
840 : }
841 :
842 5 : HcclResult MyRank::CreateChannels(CommEngine engine, const std::string &commTag,
843 : const HcclChannelDesc* channelDescs, uint32_t channelNum, ChannelHandle *channelHandles)
844 : {
845 5 : CHK_PTR_NULL(channelDescs);
846 4 : CHK_PTR_NULL(channelHandles);
847 3 : CHK_PRT_RET(channelNum == 0, HCCL_ERROR("[%s] invalid param: channelNum is zero", __func__), HCCL_E_PARA);
848 :
849 2 : HCCL_INFO("[CreateChannels][Enter] engine[%s] commTag[%s] channelNum[%u] rankId[%u]", GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), commTag.c_str(), channelNum, rankId_);
850 :
851 : // 参数检查
852 2 : CHK_RET(CheckChannelParam(engine, channelDescs, channelNum));
853 :
854 1 : std::vector<ChannelHandle> hostChannelHandles(channelNum);
855 1 : ChannelHandle *hostChannelHandleList = hostChannelHandles.data();
856 :
857 1 : auto& rdmaConfig = Hccl::EnvConfig::GetInstance().GetRdmaConfig();
858 1 : std::vector<HcommChannelDesc> hcommDescs(channelNum);
859 2 : for (u32 i = 0; i < channelNum; ++i) {
860 1 : hcommDescs[i] = MyRankUtils::ChannelDescHccl2Hcomm(channelDescs[i], config_);
861 1 : hcommDescs[i].roceAttr.qpThreshold = rdmaConfig.GetRdmaMultiQpThreshold();
862 1 : CHK_RET(ConfigSqDepthByExpansionMode(engine, hcommDescs[i]));
863 : }
864 :
865 1 : auto start = std::chrono::steady_clock::now();
866 1 : std::string socketTag = commTag + "_engine_" + std::to_string(engine);
867 1 : CHK_RET(BatchCreateSockets(channelDescs, channelNum, socketTag, hcommDescs));
868 1 : CHK_RET_UNAVAIL(BatchCreateChannels(engine, channelDescs, channelNum, hcommDescs, hostChannelHandleList));
869 :
870 1 : if (!newChannels_.empty()) {
871 0 : CHK_RET(BatchConnectChannels(channelDescs, hostChannelHandleList, channelNum));
872 0 : auto end = std::chrono::steady_clock::now();
873 0 : auto duration = std::chrono::duration_cast<std::chrono::microseconds>(end - start).count();
874 0 : HCCL_RUN_INFO("[MyRank][CreateChannels] CreateChannels Time Elapsed [%lld]us, channelNum [%u]", duration, channelNum);
875 : }
876 :
877 : // 借用hcommDescs.socket，完成一致性校验必要的数据交换
878 : DevType devType;
879 1 : CHK_RET(hrtGetDeviceType(devType));
880 1 : if (devType == DevType::DEV_TYPE_950) {
881 1 : auto startConsistency = std::chrono::steady_clock::now();
882 1 : CHK_RET(exchangeInfoMgr_.BatchExchangeAndCheckConsistency(
883 : channelDescs, hcommDescs, channelNum, newChannels_, collCommConfigConsistency_, engine));
884 0 : auto endConsistency = std::chrono::steady_clock::now();
885 0 : auto durationConsistency = std::chrono::duration_cast<std::chrono::microseconds>(endConsistency - startConsistency).count();
886 0 : HCCL_INFO("[MyRank][CreateChannels] BatchExchangeAndCheckConsistency Time Elapsed [%lld]us, channelNum [%u]",
887 : durationConsistency, channelNum);
888 : }
889 :
890 : // 添加初始化时进行填表
891 0 : for (u32 i = 0; i < channelNum; ++i) {
892 0 : u32 remoteRank = channelDescs[i].remoteRank;
893 0 : HcclCommDfx::AddChannelRemoteRankId(commTag, hostChannelHandleList[i], remoteRank);
894 : // 打印UB通道建链信息
895 0 : if (channelDescs[i].localEndpoint.loc.locType == ENDPOINT_LOC_TYPE_DEVICE &&
896 0 : channelDescs[i].remoteEndpoint.loc.locType == ENDPOINT_LOC_TYPE_DEVICE) {
897 0 : HCCL_CONFIG_DEBUG(HCCL_RES, "create channel info:channel handle[%s] comm tag[%s] protocol[%s]"
898 : " local rank[%u] local dev phyid[%u] remote rank[%u] remote dev phyid[%u] engine[%s]",
899 : std::to_string(reinterpret_cast<uint64_t>(hostChannelHandleList[i])).c_str(), commTag.c_str(),
900 : MyRankUtils::GetCommProtocolEnumStr(channelDescs[i].localEndpoint.protocol).c_str(), rankId_,
901 : channelDescs[i].localEndpoint.loc.device.devPhyId, remoteRank,
902 : channelDescs[i].remoteEndpoint.loc.device.devPhyId, GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str());
903 0 : } else {
904 0 : HCCL_CONFIG_DEBUG(HCCL_RES, "create channel info:channel handle[%s] comm tag[%s] protocol[%s]"
905 : " local rank[%u] remote rank[%u] engine[%s]",
906 : std::to_string(reinterpret_cast<uint64_t>(hostChannelHandleList[i])).c_str(), commTag.c_str(),
907 : MyRankUtils::GetCommProtocolEnumStr(channelDescs[i].localEndpoint.protocol).c_str(), rankId_,
908 : remoteRank, GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str());
909 : }
910 : }
911 :
912 0 : if (engine == COMM_ENGINE_AICPU || engine == COMM_ENGINE_AICPU_TS) {
913 : // 新增：添加 kernelLaunchAicpuCommInit 调用
914 0 : if (!callbacks_.getAicpuCommState()) {
915 0 : HCCL_INFO("MyRank::%s kernelLaunchAicpuCommInit start.", __func__);
916 0 : HcclResult ret = callbacks_.kernelLaunchAicpuCommInit();
917 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[%s] kernelLaunchAicpuCommInit failed, return [%d].", __func__, ret), ret);
918 0 : callbacks_.setAicpuCommState(true);
919 : }
920 0 : HcommChannelDesc *hcommDesc = hcommDescs.data();
921 0 : CHK_RET(ChannelProcess::ChannelKernelLaunchForComm(
922 : channelHandles, hostChannelHandleList, hcommDesc, channelNum, commTag, binHandle_));
923 :
924 : // ns recovery
925 0 : nsRecoveryProcessor_->AddNsRecoveryData(engine, channelHandles, hostChannelHandleList, channelNum, commTag);
926 :
927 0 : return HCCL_SUCCESS;
928 : }
929 :
930 0 : if (engine == COMM_ENGINE_CPU || engine == COMM_ENGINE_CCU || engine == COMM_ENGINE_AIV) {
931 : // TODO: Host侧 Channel 赋值到 channelHandles
932 0 : CHK_SAFETY_FUNC_RET(memcpy_s(channelHandles, channelNum * sizeof(ChannelHandle), hostChannelHandleList,
933 : channelNum * sizeof(ChannelHandle)));
934 0 : return HCCL_SUCCESS;
935 : }
936 :
937 0 : HCCL_ERROR("[MyRank][%s] unsupported comm engine[%s].", __func__, GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str());
938 0 : return HCCL_E_NOT_SUPPORT;
939 1 : }
940 :
941 1 : HcclResult MyRank::ChannelGetHcclBuffer(ChannelHandle channel, void **buffer, uint64_t *size)
942 : {
943 1 : CHK_PTR_NULL(buffer);
944 1 : CHK_PTR_NULL(size);
945 :
946 1 : u32 memNum = 0;
947 1 : CommMem* remoteMem = nullptr;
948 1 : char** memTags = nullptr;
949 1 : CHK_RET(static_cast<HcclResult>(HcommChannelGetRemoteMems(channel, &memNum, &remoteMem, &memTags)));
950 1 : if (memNum > 0) {
951 0 : CHK_PTR_NULL(remoteMem);
952 : // AicpuTsHccsChannel不使用memTag，返回为空，默认索引0为cclBuffer
953 0 : if (memTags == nullptr) {
954 0 : *buffer = remoteMem[0].addr;
955 0 : *size = remoteMem[0].size;
956 0 : HCCL_INFO("[%s] Found HcclBuffer : addr=%p, size=%llu", __func__, *buffer, *size);
957 0 : return HCCL_SUCCESS;
958 : }
959 0 : for (u32 i = 0; i < memNum; ++i) {
960 0 : std::string tag = memTags[i];
961 0 : if (tag == "HcclBuffer") {
962 0 : *buffer = remoteMem[i].addr;
963 0 : *size = remoteMem[i].size;
964 0 : HCCL_INFO("[%s] Found HcclBuffer : addr=%p, size=%llu", __func__, *buffer, *size);
965 0 : return HCCL_SUCCESS;
966 : }
967 0 : HCCL_INFO("[%s] Found %s : addr=%p, size=%llu", __func__, memTags[i],
968 : remoteMem[i].addr, remoteMem[i].size);
969 0 : }
970 : }
971 1 : HCCL_ERROR("[%s] HcclBuffer not found.", __func__);
972 1 : return HCCL_E_INTERNAL;
973 : }
974 :
975 4 : HcclResult MyRank::ChannelGetRemoteMems(ChannelHandle channel, uint32_t *memNum, CommMem **remoteMem, char ***memTags) const
976 : {
977 4 : CHK_PTR_NULL(remoteMem);
978 3 : CHK_PTR_NULL(memTags);
979 2 : CHK_PTR_NULL(memNum);
980 1 : CHK_RET(static_cast<HcclResult>(HcommChannelGetRemoteMems(channel, memNum, remoteMem, memTags)));
981 : // 添加空指针检查，防止返回的指针为空
982 1 : if (*memNum > 0) {
983 0 : CHK_PTR_NULL(*remoteMem);
984 0 : CHK_PTR_NULL(*memTags);
985 : }
986 1 : HCCL_INFO("[%s] success. memNum[%u]", __func__, *memNum);
987 1 : return HCCL_SUCCESS;
988 : }
989 :
990 4 : HcclResult MyRank::ChannelGetRemoteMems(ChannelHandle channel, uint32_t *memNum, CommMem **remoteMem,
991 : std::vector<std::string> &memTags) const
992 : {
993 4 : CHK_PTR_NULL(remoteMem);
994 3 : CHK_PTR_NULL(memNum);
995 2 : char **rawTags = nullptr;
996 2 : CHK_RET(static_cast<HcclResult>(HcommChannelGetRemoteMems(channel, memNum, remoteMem, &rawTags)));
997 : // 添加空指针检查，防止返回的指针为空
998 2 : if (*memNum > 0) {
999 1 : CHK_PTR_NULL(*remoteMem);
1000 1 : CHK_PTR_NULL(rawTags);
1001 1 : memTags.reserve(*memNum);
1002 3 : for (uint32_t i = 0; i < *memNum; ++i) {
1003 2 : memTags.emplace_back(rawTags[i] == nullptr ? "" : rawTags[i]);
1004 : }
1005 : }
1006 2 : HCCL_INFO("[%s] success. memNum[%u]", __func__, *memNum);
1007 2 : return HCCL_SUCCESS;
1008 : }
1009 :
1010 0 : std::vector<ChannelHandle> MyRank::GetAllChannelList()
1011 : {
1012 0 : ChannelTable channelTable = rankPairMgr_->GetChannelTable();
1013 0 : std::vector<ChannelHandle> channelList;
1014 0 : for (const auto& rankPair : channelTable) {
1015 0 : for (const auto& endPointPair : rankPair.second) {
1016 0 : for (const auto& comEngines : endPointPair.second) {
1017 0 : channelList.insert(channelList.end(), comEngines.second.begin(), comEngines.second.end());
1018 : }
1019 : }
1020 : }
1021 :
1022 0 : return channelList;
1023 0 : }
1024 :
1025 98 : void MyRank::SetKfcControlTransfer(std::shared_ptr<HDCommunicate> kfcControlTransferH2D,
1026 : std::shared_ptr<HDCommunicate> kfcStatusTransferD2H)
1027 : {
1028 98 : if (nsRecoveryProcessor_ == nullptr) {
1029 1 : HCCL_ERROR("[MyRank][SetKfcControlTransfer] nsRecoveryProcessor_ is null, cannot set KFC control transfer.");
1030 1 : return;
1031 : }
1032 97 : nsRecoveryProcessor_->SetKfcControlTransfer(kfcControlTransferH2D, kfcStatusTransferD2H);
1033 : }
1034 :
1035 0 : HcclResult MyRank::StopLaunch()
1036 : {
1037 0 : HCCL_INFO("[NsRecovery][StopLaunch] MyRank::StopLaunch start!");
1038 0 : auto ret = nsRecoveryProcessor_->StopLaunch();
1039 0 : if (ret != HcclResult::HCCL_SUCCESS) {
1040 0 : HCCL_ERROR("[NsRecovery][StopLaunch] MyRank::StopLaunch failed, ret = 0x%016llx", HCCL_ERROR_CODE(ret));
1041 : }
1042 0 : HCCL_INFO("[NsRecovery][StopLaunch] MyRank::StopLaunch success!");
1043 0 : return ret;
1044 : }
1045 :
1046 0 : HcclResult MyRank::Clean()
1047 : {
1048 0 : HCCL_INFO("[NsRecovery][Clean] MyRank::Clean start!");
1049 0 : auto channelList = GetAllChannelList();
1050 0 : if (channelList.empty()) {
1051 0 : HCCL_INFO("[NsRecovery][Clean] Channel list empty, No need to clean!");
1052 0 : return HcclResult::HCCL_SUCCESS;
1053 : }
1054 0 : auto ret = ChannelProcess::ChannelClean(channelList.data(), channelList.size());
1055 0 : if (ret != HcclResult::HCCL_SUCCESS) {
1056 0 : HCCL_ERROR("[NsRecovery][Clean] MyRank::Clean failed, ret = 0x%016llx", HCCL_ERROR_CODE(ret));
1057 0 : return ret;
1058 : }
1059 :
1060 0 : ret = nsRecoveryProcessor_->Clean();
1061 0 : if (ret != HcclResult::HCCL_SUCCESS) {
1062 0 : HCCL_ERROR("[NsRecovery][Clean] MyRank::Clean failed, ret = 0x%016llx", HCCL_ERROR_CODE(ret));
1063 0 : return ret;
1064 : }
1065 :
1066 0 : HCCL_INFO("[NsRecovery][Clean] MyRank::Clean success!");
1067 0 : return HcclResult::HCCL_SUCCESS;
1068 0 : }
1069 :
1070 0 : HcclResult MyRank::Resume()
1071 : {
1072 0 : HCCL_INFO("[NsRecovery][Resume] MyRank::Resume start!");
1073 0 : auto channelList = GetAllChannelList();
1074 0 : if (channelList.empty()) {
1075 0 : HCCL_INFO("[NsRecovery][Resume] Resume list empty, No need to resume!");
1076 0 : return HcclResult::HCCL_SUCCESS;
1077 : }
1078 :
1079 0 : auto ret = ChannelProcess::ChannelResume(channelList.data(), channelList.size());
1080 0 : if (ret != HcclResult::HCCL_SUCCESS) {
1081 0 : HCCL_ERROR("[NsRecovery][Resume] MyRank::Resume failed, ret = 0x%016llx", HCCL_ERROR_CODE(ret));
1082 0 : return ret;
1083 : }
1084 :
1085 0 : ret = nsRecoveryProcessor_->Resume(binHandle_);
1086 0 : if (ret != HcclResult::HCCL_SUCCESS) {
1087 0 : HCCL_ERROR("[NsRecovery][Resume] MyRank::Resume failed, ret = 0x%016llx", HCCL_ERROR_CODE(ret));
1088 0 : return ret;
1089 : }
1090 :
1091 0 : HCCL_INFO("[NsRecovery][Resume] MyRank::Resume success!");
1092 0 : return HCCL_SUCCESS;
1093 0 : }
1094 :
1095 7 : CollCommConfigConsistency &MyRank::GetCollCommConfigConsistency()
1096 : {
1097 7 : return collCommConfigConsistency_;
1098 : }
1099 :
1100 : } // namespace hccl
|
