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