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