Line data Source code
1 : /**
2 : * Copyright (c) 2025 Huawei Technologies Co., Ltd.
3 : * This program is free software, you can redistribute it and/or modify it under the terms and conditions of
4 : * CANN Open Software License Agreement Version 2.0 (the "License").
5 : * Please refer to the License for details. You may not use this file except in compliance with the License.
6 : * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,
7 : * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
8 : * See LICENSE in the root of the software repository for the full text of the License.
9 : */
10 :
11 : #include "my_rank.h"
12 : #include "hcomm_c_adpt.h"
13 : #include "endpoint_pair.h"
14 : #include "hccl_res.h"
15 : #include "../common/loggers/channel_logger.h" // 日志记录器
16 : #include "hcclCommDfx.h"
17 : #include "config/env_config.h"
18 : #include "env_config/env_config_v2.h"
19 : #include "channel_process.h"
20 : #include "ccu_dev_mgr_imp.h"
21 : #include "ccu_device_res.h"
22 : #include "ccu_res_desc.h"
23 : #include "ccu_device_pub.h"
24 : #include "ccu_res_desc_mgr.h"
25 : #include "ccu_log.h"
26 : #include "dlprof_function.h"
27 : #include "config_log.h"
28 : #include "comm_engine_utils.h"
29 : #include "hcom_common.h"
30 : #include "op_base.h"
31 : #include "ccu_res.h"
32 : #include "coll_comm_mgr.h"
33 : #include "new_rank_info.h"
34 :
35 : #include <acl/acl.h>
36 : #include "shared_jetty_channel_pool.h"
37 :
38 : using namespace hcomm;
39 :
40 : namespace MyRankUtils {
41 :
42 29 : uint32_t ResolveUbCommDomainQos(const hccl::CommConfig& commConfig)
43 : {
44 29 : if (commConfig.GetConfigHcclQos() == HCCL_COMM_QOS_CONFIG_NOT_SET) {
45 25 : return EnvConfig::UB_QOS_DEFAULT;
46 : }
47 4 : return commConfig.GetConfigHcclQos();
48 : }
49 :
50 37 : HcommChannelDesc ChannelDescHccl2Hcomm(const HcclChannelDesc& hcclDesc, const hccl::CommConfig& commConfig)
51 : {
52 37 : HcommChannelDesc hcommDesc{};
53 37 : (void)HcommChannelDescInit(&hcommDesc, 1);
54 37 : hcommDesc.remoteEndpoint = hcclDesc.remoteEndpoint;
55 37 : hcommDesc.notifyNum = hcclDesc.notifyNum;
56 37 : hcommDesc.memHandles = reinterpret_cast<HcommMemHandle*>(hcclDesc.memHandles);
57 37 : hcommDesc.memHandleNum = hcclDesc.memHandleNum;
58 37 : (void)memcpy_s(hcommDesc.raws, sizeof(hcommDesc.raws), hcclDesc.raws, sizeof(hcommDesc.raws));
59 : // RoCE：透传原始 hcclQos（可为 NOT_SET），由 CheckRoceAttr/ApplyRoceQosCompatToSlTc 决定是否映射 SL/TC
60 37 : if (hcclDesc.channelProtocol == COMM_PROTOCOL_ROCE) {
61 8 : hcommDesc.qos = commConfig.GetConfigHcclQos();
62 8 : hcommDesc.roceAttr.retryCnt = hcclDesc.roceAttr.retryCnt;
63 8 : hcommDesc.roceAttr.retryInterval = hcclDesc.roceAttr.retryInterval;
64 8 : hcommDesc.roceAttr.sl = hcclDesc.roceAttr.sl;
65 8 : hcommDesc.roceAttr.tc = hcclDesc.roceAttr.tc;
66 8 : return hcommDesc;
67 : }
68 : // UB 等：未配置时落默认 4，供下游 Jetty/TP 使用
69 29 : hcommDesc.qos = ResolveUbCommDomainQos(commConfig);
70 29 : if (hcclDesc.channelProtocol == COMM_PROTOCOL_UB_MEM) {
71 16 : hcommDesc.ubMemAttr.pathMode = hcclDesc.ubMemAttr.pathMode;
72 : }
73 29 : return hcommDesc;
74 : }
75 :
76 : /* 公共模块函数返回值定义，跟业务层同步 */
77 : const std::unordered_map<CommProtocol, std::string> HCOM_COMM_PROTOCOL_STR_MAP
78 : = {{COMM_PROTOCOL_RESERVED, "RESERVED"}, {COMM_PROTOCOL_HCCS, "HCCS"}, {COMM_PROTOCOL_ROCE, "ROCE"},
79 : {COMM_PROTOCOL_PCIE, "PCIE"}, {COMM_PROTOCOL_SIO, "SIO"}, {COMM_PROTOCOL_UB_CTP, "UB_CTP"},
80 : {COMM_PROTOCOL_UBC_TP, "UBC_TP"}, {COMM_PROTOCOL_UB_MEM, "UB_MEM"}, {COMM_PROTOCOL_UBOE, "UBOE"},
81 : {COMM_PROTOCOL_UB_RTP, "UB_RTP"}};
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 0 : inline const char* GetTlsTypeStr(EndpointLocType localType)
94 : {
95 0 : return localType == ENDPOINT_LOC_TYPE_HOST ? "HostDpu" : "Device";
96 : }
97 :
98 0 : inline const char* GetTlsStatusStr(Hccl::TlsStatus tlsStatus)
99 : {
100 0 : if (tlsStatus == Hccl::TlsStatus::ENABLE) {
101 0 : return "ENABLE";
102 : }
103 0 : if (tlsStatus == Hccl::TlsStatus::DISABLE) {
104 0 : return "DISABLE";
105 : }
106 0 : return "UNKNOWN";
107 : }
108 :
109 : } // namespace MyRankUtils
110 :
111 5 : HcclResult MyRankUtils::FillRoceSrcPortList(
112 : const HcclChannelDesc& hcclDesc, HcommChannelDesc& hcommDesc, std::vector<uint16_t>& srcPortBuf)
113 : {
114 5 : hcommDesc.roceAttr.srcPortList = nullptr;
115 5 : if (hcommDesc.remoteEndpoint.protocol != COMM_PROTOCOL_ROCE || hcommDesc.exchangeAllMems) {
116 3 : HCCL_INFO(
117 : "[%s] skip: protocol[%d] exchangeAllMems[%d]", __func__, hcommDesc.remoteEndpoint.protocol,
118 : hcommDesc.exchangeAllMems);
119 3 : return HCCL_SUCCESS;
120 : }
121 2 : const auto& qpSrcPortConfig = Hccl::EnvConfig::GetInstance().GetRdmaConfig().GetMultiQpSrcPortConfig();
122 2 : if (!qpSrcPortConfig.IsAvailable()) {
123 0 : HCCL_INFO(
124 : "[%s] skip: multiQpSrcPortConfig not available (env HCCL_RDMA_QP_PORT_CONFIG_PATH unset or "
125 : "MultiQpSrcPort.cfg empty)",
126 : __func__);
127 0 : return HCCL_SUCCESS;
128 : }
129 2 : Hccl::IpAddress localIp;
130 2 : Hccl::IpAddress remoteIp;
131 2 : HcclResult localRet = CommAddrToIpAddress(hcclDesc.localEndpoint.commAddr, localIp);
132 2 : HcclResult remoteRet = CommAddrToIpAddress(hcclDesc.remoteEndpoint.commAddr, remoteIp);
133 2 : CHK_PRT_RET(
134 : localRet != HCCL_SUCCESS || remoteRet != HCCL_SUCCESS,
135 : HCCL_ERROR("[%s] CommAddrToIpAddress failed: localRet[%d] remoteRet[%d]", __func__, localRet, remoteRet),
136 : HCCL_E_INTERNAL);
137 1 : auto ports = Hccl::GetMultiQpSrcPortsByIpPair(qpSrcPortConfig, localIp, remoteIp);
138 1 : if (ports.empty()) {
139 0 : HCCL_INFO(
140 : "[%s] skip: no matching ports for localIp[%s] remoteIp[%s]", __func__, localIp.GetIpStr().c_str(),
141 : remoteIp.GetIpStr().c_str());
142 0 : return HCCL_SUCCESS;
143 : }
144 1 : u32 queueNum = hcommDesc.roceAttr.queueNum;
145 1 : srcPortBuf.resize(queueNum);
146 5 : for (u32 j = 0; j < queueNum; ++j) {
147 4 : srcPortBuf[j] = ports[j % ports.size()];
148 : }
149 1 : hcommDesc.roceAttr.srcPortList = srcPortBuf.data();
150 1 : HCCL_INFO("[%s] success: queueNum[%u] portCount[%zu]", __func__, queueNum, ports.size());
151 1 : return HCCL_SUCCESS;
152 1 : }
153 :
154 : namespace hccl {
155 :
156 : constexpr uint32_t UNREUSE_CHANNEL_IDX = 0xFFFFFFFF;
157 :
158 286 : MyRank::MyRank(
159 : aclrtBinHandle binHandle, uint32_t rankId, const CommConfig& config, const ManagerCallbacks& callbacks,
160 286 : RankGraph* rankGraph, const Hccl::RankIpPortMapPtr& rankIpPortMap)
161 286 : : binHandle_(binHandle),
162 286 : rankId_(rankId),
163 286 : config_(config),
164 286 : callbacks_(callbacks),
165 286 : rankGraph_(rankGraph),
166 286 : rankIpPortMap_(rankIpPortMap)
167 286 : {}
168 :
169 572 : MyRank::~MyRank()
170 : {
171 286 : HCCL_INFO("[MyRank][~MyRank] MyRank deinit, rankId_[%u], devLogicId_[%d]", rankId_, devLogicId_);
172 : // 共享 Jetty Channel 不归 rankPairMgr_ 管理，需在 rankPairMgr_ 析构前独立清理
173 286 : (void)SharedJettyChannelPool::GetInstance().DestroyAllByMyRank(this);
174 : // 先清空反查索引，避免 rankPairMgr_ 析构 EndpointPair 时仍持有指向其的裸指针；
175 : // 持锁保证与并发 DestroyChannels 的索引读写一致
176 : {
177 286 : std::lock_guard<std::mutex> lock(channelIndexMtx_);
178 286 : handleToEpPair_.clear();
179 286 : }
180 : // 析构有时序要求
181 286 : rankPairMgr_ = nullptr; // 内部会销毁channel，可能需要返还endpoint与ccu资源
182 286 : endpointMgr_ = nullptr; // 内部会销毁endpoint，可能需要返回ccu资源
183 :
184 : struct ResourceCleanupGuard {
185 286 : explicit ResourceCleanupGuard(MyRank& myRank) : myRank_(myRank) {}
186 286 : ~ResourceCleanupGuard() noexcept
187 : {
188 286 : myRank_.ccuInsHandle_ = 0;
189 286 : myRank_.assignedCcuInsHandle_ = 0;
190 :
191 286 : if (myRank_.ccuDrvHandle_) {
192 1 : myRank_.ccuDrvHandle_ = nullptr; // 先减少引用计数，再尝试关闭
193 1 : (void)CcuDeinitFeature(myRank_.devLogicId_);
194 : // 尝试关闭CCU功能，最后一个调用时会关闭CCU驱动
195 : }
196 :
197 286 : myRank_.ReleaseCcuMsCommReservation();
198 :
199 286 : myRank_.commMems_ = nullptr;
200 286 : myRank_.nsRecoveryProcessor_ = nullptr;
201 286 : }
202 :
203 : MyRank& myRank_;
204 286 : } cleanupGuard(*this);
205 :
206 286 : if (ccuInsHandle_ != 0 || assignedCcuInsHandle_ != 0) { // 内部清理CCU资源，关闭CCU通道
207 : // 刷新并获取当前线程的 DeviceId
208 13 : int32_t threadDevId = INVALID_INT;
209 13 : CHK_RET_NULL(HcclDeviceRefresh(threadDevId));
210 13 : HCCL_INFO("[%s] curDeviceLogicId[%d], threadDevId[%d]", __func__, devLogicId_, threadDevId);
211 : // 先切换为目标 curDeviceLogicId
212 13 : bool isDiffDevId = false;
213 13 : if (devLogicId_ != threadDevId) {
214 0 : CHK_RET_NULL(hrtSetDevice(devLogicId_));
215 0 : isDiffDevId = true;
216 : }
217 : // 销毁通信域自有的 ccuInstance（QueryCcuIns 创建）
218 13 : if (ccuInsHandle_ != 0) {
219 6 : CHK_PRT(static_cast<HcclResult>(HcommCcuInsDestroy(ccuInsHandle_)));
220 : }
221 : // 销毁通过 Assign 绑定的 ccuInstance（所有权已转移给通信域）
222 13 : if (assignedCcuInsHandle_ != 0 && assignedCcuInsHandle_ != ccuInsHandle_) {
223 8 : CHK_PRT(static_cast<HcclResult>(HcommCcuInsDestroy(assignedCcuInsHandle_)));
224 : }
225 : // 切换回原来的 DeviceId
226 13 : if (isDiffDevId) {
227 0 : CHK_RET_NULL(hrtSetDevice(threadDevId));
228 0 : CHK_PRT(HcclDeviceRefresh(threadDevId));
229 : }
230 : }
231 572 : }
232 :
233 10 : HcclResult MyRank::GetLocalTlsStatus(EndpointLocType localType, Hccl::TlsStatus& tlsStatus) const
234 : {
235 10 : tlsStatus = Hccl::TlsStatus::UNKNOWN;
236 10 : s32 deviceLogicId = -1;
237 10 : u32 devicePhyId = INVALID_UINT;
238 10 : CHK_RET(hrtGetDevice(&deviceLogicId));
239 9 : CHK_RET(hrtGetDevicePhyIdByIndex(static_cast<u32>(deviceLogicId), devicePhyId));
240 :
241 8 : RaInfo info{};
242 8 : info.mode = localType == ENDPOINT_LOC_TYPE_HOST ? NetworkMode::NETWORK_PEER_ONLINE : NetworkMode::NETWORK_OFFLINE;
243 8 : info.phyId = devicePhyId;
244 8 : return Hccl::HrtRaGetTlsStatus(&info, tlsStatus);
245 : }
246 :
247 4 : void MyRank::GetAbnormalChannelTlsStatus(
248 : const HcclChannelDesc* channelDescs, const int32_t* statusList, uint32_t channelNum,
249 : std::vector<Hccl::TlsStatus>& tlsStatusList) const
250 : {
251 4 : tlsStatusList.assign(channelNum, Hccl::TlsStatus::UNKNOWN);
252 14 : for (uint32_t i = 0; i < channelNum; ++i) {
253 10 : if (statusList[i] == ChannelStatus::READY) {
254 3 : continue;
255 : }
256 7 : Hccl::TlsStatus tlsStatus = Hccl::TlsStatus::UNKNOWN;
257 7 : HcclResult ret = GetLocalTlsStatus(channelDescs[i].localEndpoint.loc.locType, tlsStatus);
258 7 : if (ret != HCCL_SUCCESS) {
259 0 : HCCL_WARNING("[GetLocalTlsStatus] Can not get TlsStatus, channelIndex[%u], ret[%d]", i, ret);
260 0 : continue;
261 : }
262 7 : tlsStatusList[i] = tlsStatus;
263 : }
264 4 : }
265 :
266 37 : HcclResult MyRank::RegisterCommMemsToEndpoint(EndpointHandle epHandle)
267 : {
268 37 : std::vector<HcclMem> memVec;
269 37 : std::vector<std::string> memTag;
270 37 : uint64_t version = 0;
271 37 : CHK_RET(commMems_->GetAllMemory(memVec, memTag, version));
272 37 : HCCL_INFO("[%s] got %zu memory regions to register, version[%llu]", __func__, memVec.size(), version);
273 37 : CHK_RET(endpointMgr_->RegisterMemory(epHandle, memTag, memVec, version));
274 36 : return HCCL_SUCCESS;
275 37 : }
276 :
277 37 : HcclResult MyRank::PrepareMemHandles(
278 : EndpointHandle epHandle, void** memHandles, uint32_t memHandleNum, std::vector<MemHandle>& memHandleVec)
279 : {
280 : // 从 CommMems 提取该 channel 需要的 tag 列表
281 : // GetTagsFromHandles 始终 push cclBuffer；用户 handles 异常时内部跳过，不阻断注册
282 37 : std::vector<std::string> memTags;
283 37 : CHK_RET(commMems_->GetTagsFromHandles(memHandles, memHandleNum, memTags));
284 :
285 : // 确保 CommMems 全量内存已注册到该 endpoint（版本一致则跳过）
286 37 : CHK_RET(RegisterCommMemsToEndpoint(epHandle));
287 :
288 : // 从 endpoint 查询指定 tag 的 MemHandle
289 36 : CHK_RET(endpointMgr_->GetMemHandlesByTags(epHandle, memTags, memHandleVec));
290 36 : return HCCL_SUCCESS;
291 37 : }
292 :
293 1 : HcclResult MyRank::UnregMemByTag(const std::string& tag)
294 : {
295 1 : CHK_PTR_NULL(endpointMgr_);
296 1 : return endpointMgr_->UnregMemByTag(tag);
297 : }
298 :
299 7 : HcclResult MyRank::ReserveCcuMsCommOrFallback()
300 : {
301 7 : if (opExpansionMode_ != CCU_MS_MODE) {
302 2 : return HCCL_SUCCESS;
303 : }
304 :
305 5 : bool reserved = false;
306 5 : CHK_RET(CollCommMgr::GetInstance().TryReserveCcuMsComm(devLogicId_, config_.GetConfigCommName(), reserved));
307 5 : if (reserved) {
308 4 : ccuMsCommReserved_ = true;
309 4 : return HCCL_SUCCESS;
310 : }
311 :
312 1 : opExpansionMode_ = CCU_SCHED_MODE;
313 1 : HCCL_RUN_WARNING(
314 : "[MyRank][%s] CCU_MS comm already exists on device[%d], fallback to CCU_SCHED, rankId[%u].", __func__,
315 : devLogicId_, rankId_);
316 1 : return HCCL_SUCCESS;
317 : }
318 :
319 293 : void MyRank::ReleaseCcuMsCommReservation()
320 : {
321 293 : if (!ccuMsCommReserved_) {
322 288 : return;
323 : }
324 5 : CollCommMgr::GetInstance().ReleaseCcuMsComm(devLogicId_, config_.GetConfigCommName());
325 5 : ccuMsCommReserved_ = false;
326 : }
327 :
328 7 : void MyRank::ReconcileCcuMsCommReservation(HcclResult initRet)
329 : {
330 7 : if (initRet != HCCL_SUCCESS || opExpansionMode_ != CCU_MS_MODE) {
331 7 : ReleaseCcuMsCommReservation();
332 : }
333 7 : }
334 :
335 7 : HcclResult MyRank::TryInitCcuInstanceOnDemand()
336 : {
337 7 : auto ccuInsType = OpExpansionModeToCcuInstanceType(opExpansionMode_);
338 7 : if (ccuInsType == CcuInstanceType::CCU_UNUSED) {
339 2 : ccuInsHandle_ = 0;
340 2 : return HcclResult::HCCL_SUCCESS;
341 : }
342 :
343 5 : if (mainBoardType_ == Hccl::HcclMainboardId::MAINBOARD_OTHERS) {
344 5 : CHK_RET(CcuGetMainboardType(devLogicId_, mainBoardType_));
345 : }
346 :
347 5 : if (mainBoardType_ == Hccl::HcclMainboardId::MAINBOARD_PCIE_STD
348 5 : && ccuInsType == CcuInstanceType::CCU_MS) { // 标卡环境下配置CCU_MS拦截报错
349 0 : HCCL_ERROR(
350 : "[%s] ccuInstanceType[%d] not support in %s", __func__, ccuInsType, mainBoardType_.Describe().c_str());
351 0 : return HcclResult::HCCL_E_NOT_SUPPORT;
352 : }
353 :
354 : // 拉起ccu驱动
355 5 : if (!ccuDrvHandle_) {
356 5 : auto ccuInitRet = CcuInitFeature(devLogicId_, ccuDrvHandle_);
357 : // ccu驱动拉起失败，直接回退至aicpu ts
358 5 : if (ccuInitRet == CcuResult::CCU_E_DRV_BUSY) {
359 2 : opExpansionMode_ = AICPU_TS_MODE;
360 2 : ccuInsHandle_ = 0;
361 2 : HCCL_RUN_WARNING(
362 : "[MyRank][%s] failed to init ccu driver, "
363 : "fallback to aicpu, rankId[%u].",
364 : __func__, rankId_);
365 2 : return HcclResult::HCCL_SUCCESS;
366 : }
367 :
368 : // 预期外返回值属于错误
369 3 : if (ccuInitRet != CcuResult::CCU_SUCCESS) {
370 2 : HCCL_ERROR("[%s] failed, ret[%d] is not expected.", __func__, ccuInitRet);
371 2 : ccuInsHandle_ = 0;
372 2 : return static_cast<HcclResult>(ccuInitRet);
373 : }
374 : }
375 :
376 : // ccu驱动拉起成功
377 1 : return HcclResult::HCCL_SUCCESS;
378 : }
379 :
380 7 : HcclResult MyRank::TryInitCcuInstance()
381 : {
382 7 : CHK_RET(ReserveCcuMsCommOrFallback());
383 :
384 : // ccu instance 不在 init 时创建，由 QueryCcuIns 创建或由 Assign 显式绑定；
385 : // 此处仅拉起 ccu 驱动。
386 7 : HcclResult ret = TryInitCcuInstanceOnDemand();
387 7 : ReconcileCcuMsCommReservation(ret);
388 7 : return ret;
389 : }
390 :
391 77 : HcclResult MyRank::GetDevicePortInternal(uint32_t rank, uint32_t* devPort, EndpointLocType locType)
392 : {
393 77 : CHK_PTR_NULL(devPort);
394 77 : CHK_PTR_NULL(rankGraph_);
395 :
396 : DevType devType;
397 77 : CHK_RET(hrtGetDeviceType(devType));
398 : // v1 模式 (mode_ == 0): 强制转换为 RankGraphV1 调用 GetDevicePort
399 : // v2 模式 (mode_ != 0): 使用 rankGraph_->GetDevicePort()
400 77 : if (devType == DevType::DEV_TYPE_910B) {
401 0 : RankGraphV1* rankGraphV1 = static_cast<RankGraphV1*>(rankGraph_);
402 0 : CHK_RET(rankGraphV1->GetDevicePort(rank, devPort));
403 : } else {
404 77 : CHK_RET(rankGraph_->GetListenPort(rank, devPort, locType));
405 : }
406 77 : return HCCL_SUCCESS;
407 : }
408 :
409 204 : HcclResult MyRank::Init(HcclMem cclBuffer, const uint32_t opExpansionMode, uint32_t rankNum)
410 : {
411 : // EXCEPTION_HANDLE_BEGIN
412 204 : CHK_RET(hrtGetDevice(&devLogicId_));
413 :
414 : // ns recovery processor初始化
415 204 : EXCEPTION_CATCH(nsRecoveryProcessor_ = std::make_unique<NsRecoveryProcessor>(), return HCCL_E_PTR);
416 :
417 : // 创建通信内存管理器
418 204 : EXCEPTION_CATCH(commMems_ = std::make_unique<CommMems>(config_.GetConfigBufferSize()), return HCCL_E_PTR);
419 :
420 : // 初始化通信内存
421 204 : CHK_RET(commMems_->Init(cclBuffer));
422 :
423 204 : EXCEPTION_CATCH(engineCtxs_ = std::make_unique<EngineCtxs>(), return HCCL_E_PTR);
424 :
425 : // 通信域配置config优先级更高，当配置默认展开模式时，读取环境变量配置
426 204 : opExpansionMode_ = opExpansionMode;
427 204 : 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 204 : 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 203 : EXCEPTION_CATCH(endpointMgr_ = std::make_unique<hcomm::EndpointMgr>(), return HCCL_E_PTR);
447 :
448 : // rankPairMgr_初始化
449 203 : EXCEPTION_CATCH(rankPairMgr_ = std::make_unique<RankPairMgr>(rankIpPortMap_), return HCCL_E_PTR);
450 :
451 203 : DlProfFunction::GetInstance().DlProfFunctionInit();
452 : // EXCEPTION_HANDLE_END
453 203 : 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(
465 : "[%s] Invalid port[%u] of Rank[%u], max valid port is %u", __func__, rmtPort, remoteRank,
466 : Hccl::MAX_VALUE_TCPPORT);
467 1 : return HCCL_E_PARA;
468 : }
469 : // 查询该socket链接的server端监听的端口（监听方的选择策略需要跟SocketConfig中保持一致）
470 22 : Hccl::IpAddress localIpAddr{};
471 22 : Hccl::IpAddress remoteIpAddr{};
472 22 : CHK_RET(CommAddrToIpAddress(localEndpointDesc.commAddr, localIpAddr));
473 22 : CHK_RET(CommAddrToIpAddress(remoteEndpointDesc.commAddr, remoteIpAddr));
474 22 : if (localIpAddr < remoteIpAddr) {
475 : // 查询localRankId对应的devPort
476 20 : CHK_RET(GetDevicePortInternal(localRank, &listenPort, localEndpointDesc.loc.locType));
477 20 : hcommDesc.role = HcommSocketRole::HCOMM_SOCKET_ROLE_SERVER;
478 20 : if (listenPort > Hccl::MAX_VALUE_TCPPORT) {
479 0 : HCCL_ERROR("[%s] Invalid port[%u] of Rank[%u]", __func__, listenPort, localRank);
480 0 : return HCCL_E_PARA;
481 : }
482 20 : hcommDesc.port = static_cast<uint16_t>(listenPort); // HcommChannelDesc.port中填监听端口号
483 : } else {
484 2 : listenPort = rmtPort;
485 2 : hcommDesc.role = HcommSocketRole::HCOMM_SOCKET_ROLE_CLIENT;
486 : hcommDesc.port
487 2 : = static_cast<uint16_t>(rmtPort); // HcommChannelDesc.port中填对端端口号(此场景下对端端口号也就是监听端口号)
488 : }
489 :
490 22 : return HCCL_SUCCESS;
491 : }
492 :
493 40 : HcclResult MyRank::GetEndpointPairFromChannel(
494 : const HcclChannelDesc& channelDesc, uint32_t channelIndex, uint32_t channelNum, uint32_t& remoteRank,
495 : hcomm::EndpointPair*& endpointPair, RankPair*& rankPair)
496 : {
497 40 : remoteRank = channelDesc.remoteRank;
498 40 : HCCL_INFO(
499 : "[%s][%u/%u] remoteRank[%u] localProtocol[%d] remoteProtocol[%d]", __func__, channelIndex + 1, channelNum,
500 : remoteRank, channelDesc.localEndpoint.protocol, channelDesc.remoteEndpoint.protocol);
501 :
502 40 : const RankIdPair rankIdPair = std::make_pair(rankId_, remoteRank);
503 40 : const EndpointDescPair endpointDescPair = std::make_pair(channelDesc.localEndpoint, channelDesc.remoteEndpoint);
504 40 : CHK_RET(rankPairMgr_->Get(rankIdPair, rankPair));
505 40 : CHK_PTR_NULL(rankPair);
506 40 : CHK_RET(rankPair->GetEndpointPair(endpointDescPair, endpointPair));
507 40 : CHK_PTR_NULL(endpointPair);
508 40 : return HCCL_SUCCESS;
509 : }
510 :
511 40 : inline std::string AddProtocolToSocketTag(const std::string& socketTag, const HcclChannelDesc* channelDescs)
512 : {
513 40 : std::string newSocketTag = socketTag + "_protocol_" + std::to_string(channelDescs->channelProtocol);
514 40 : return newSocketTag;
515 : }
516 :
517 8 : HcclResult MyRank::BatchServerInitForChannels(
518 : const HcclChannelDesc* channelDescs, uint32_t channelNum, const std::string& socketTag,
519 : ReuseSocketIdxMap& reuseSocketIdxMap)
520 : {
521 : // 批量获取socket，与server监听隔离开
522 28 : for (uint32_t i = 0; i < channelNum; ++i) {
523 20 : hcomm::EndpointPair* endpointPair = nullptr;
524 20 : RankPair* rankPair = nullptr;
525 20 : uint32_t remoteRank = 0;
526 :
527 20 : CHK_RET(GetEndpointPairFromChannel(channelDescs[i], i, channelNum, remoteRank, endpointPair, rankPair));
528 :
529 20 : if (reuseSocketIdxMap.find(rankPair) == reuseSocketIdxMap.end()) {
530 11 : std::unordered_map<hcomm::EndpointPair*, u32> endpointPair2Idx{};
531 11 : endpointPair2Idx.emplace(endpointPair, 0);
532 11 : reuseSocketIdxMap.emplace(rankPair, endpointPair2Idx);
533 20 : } else if (reuseSocketIdxMap[rankPair].find(endpointPair) == reuseSocketIdxMap[rankPair].end()) {
534 3 : reuseSocketIdxMap[rankPair].emplace(endpointPair, 0);
535 : }
536 20 : u32& reuseIdx = reuseSocketIdxMap[rankPair][endpointPair];
537 :
538 : uint32_t devicePhyId;
539 : uint32_t remoteDevicePhyId;
540 20 : rankGraph_->GetDeviceId(rankId_, &devicePhyId);
541 20 : rankGraph_->GetDeviceId(remoteRank, &remoteDevicePhyId);
542 :
543 20 : const std::string socketTagAddProto = AddProtocolToSocketTag(socketTag, &channelDescs[i]);
544 20 : auto ret = endpointPair->ServerInit(
545 : rankId_, remoteRank, socketTagAddProto, reuseIdx, devicePhyId, remoteDevicePhyId);
546 20 : CHK_PRT_RET(
547 : ret != HCCL_SUCCESS,
548 : HCCL_ERROR(
549 : "[%s] ServerInitFailed, channelIndex[%u], remoteRank[%u], protocol[%d] reuseIdx[%u]", __func__, i,
550 : remoteRank, channelDescs[i].localEndpoint.protocol, reuseIdx),
551 : ret);
552 :
553 20 : HCCL_INFO(
554 : "[%s][%u/%u] server listen successfully, remoteRank[%u], reuseIdx[%u]", __func__, i + 1, channelNum,
555 : remoteRank, reuseIdx);
556 20 : }
557 8 : return HCCL_SUCCESS;
558 : }
559 :
560 8 : HcclResult MyRank::BatchGetSocketsForChannels(
561 : const HcclChannelDesc* channelDescs, uint32_t channelNum, const std::string& socketTag,
562 : std::vector<HcommChannelDesc>& hcommDescs, ReuseSocketIdxMap& reuseSocketIdxMap)
563 : {
564 28 : for (uint32_t i = 0; i < channelNum; ++i) {
565 20 : hcomm::EndpointPair* endpointPair = nullptr;
566 20 : RankPair* rankPair = nullptr;
567 20 : uint32_t remoteRank = 0;
568 :
569 20 : CHK_RET(GetEndpointPairFromChannel(channelDescs[i], i, channelNum, remoteRank, endpointPair, rankPair));
570 :
571 20 : uint32_t listenPort = 0;
572 20 : CHK_RET(QueryListenPort(
573 : rankId_, remoteRank, channelDescs[i].localEndpoint, channelDescs[i].remoteEndpoint, listenPort,
574 : hcommDescs[i]));
575 :
576 20 : u32& reuseIdx = reuseSocketIdxMap[rankPair][endpointPair];
577 : uint32_t devicePhyId;
578 : uint32_t remoteDevicePhyId;
579 20 : rankGraph_->GetDeviceId(rankId_, &devicePhyId);
580 20 : rankGraph_->GetDeviceId(remoteRank, &remoteDevicePhyId);
581 20 : HCCL_INFO(
582 : "[MyRank][BatchCreateSockets] rankId_[%u] devicePhyId[%u] remoteRank[%u] remoteDevicePhyId[%u]", rankId_,
583 : devicePhyId, remoteRank, remoteDevicePhyId);
584 20 : Hccl::Socket* socket = nullptr;
585 20 : const std::string socketTagAddProto = AddProtocolToSocketTag(socketTag, &channelDescs[i]);
586 20 : auto ret = endpointPair->GetConnectedSocket(
587 : rankId_, remoteRank, socketTagAddProto, reuseIdx, listenPort, socket, devicePhyId, remoteDevicePhyId);
588 20 : CHK_PRT_RET(
589 : ret != HCCL_SUCCESS,
590 : HCCL_ERROR(
591 : "[%s] failed to get socket, channelIndex[%u], remoteRank[%u], protocol[%d], reuseIdx[%u], tag[%s]",
592 : __func__, i, remoteRank, channelDescs[i].localEndpoint.protocol, reuseIdx, socketTagAddProto.c_str()),
593 : ret);
594 20 : CHK_PTR_NULL(socket);
595 :
596 20 : hcommDescs[i].socket = reinterpret_cast<HcommSocket>(socket);
597 :
598 20 : HCCL_INFO(
599 : "[%s][%u/%u] socket created successfully, remoteRank[%u], socket[%p] reuseIdx[%u]", __func__, i + 1,
600 : channelNum, remoteRank, socket, reuseIdx);
601 20 : reuseIdx++;
602 20 : }
603 8 : return HCCL_SUCCESS;
604 : }
605 :
606 8 : HcclResult MyRank::BatchCreateSockets(
607 : const HcclChannelDesc* channelDescs, uint32_t channelNum, const std::string& socketTag,
608 : std::vector<HcommChannelDesc>& hcommDescs)
609 : {
610 8 : CHK_PTR_NULL(channelDescs);
611 8 : CHK_PRT_RET(channelNum == 0, HCCL_ERROR("[%s] invalid param: channelNum is zero", __func__), HCCL_E_PARA);
612 :
613 8 : ReuseSocketIdxMap reuseSocketIdxMap{};
614 : // socket服务器首先监听
615 8 : CHK_RET(BatchServerInitForChannels(channelDescs, channelNum, socketTag, reuseSocketIdxMap));
616 : // socket添加白名单以及进行连接，获取最后的socket
617 8 : CHK_RET(BatchGetSocketsForChannels(channelDescs, channelNum, socketTag, hcommDescs, reuseSocketIdxMap));
618 8 : return HCCL_SUCCESS;
619 8 : }
620 :
621 1 : HcclResult MyRank::BatchExchangeAndCheckConsistency(
622 : const HcclChannelDesc* channelDescs, const std::vector<HcommChannelDesc>& hcommDescs, uint32_t channelNum,
623 : const std::vector<std::pair<u32, u32>>& newChannels, CommEngine engine)
624 : {
625 1 : CHK_PTR_NULL(channelDescs);
626 1 : CHK_PRT_RET(channelNum == 0, HCCL_ERROR("[%s] invalid param: channelNum is zero", __func__), HCCL_E_PARA);
627 :
628 : // 与非共享路径 MyRank::CreateChannels 一致：仅 DEV_TYPE_950 需要执行通信域一致性校验交换。
629 : DevType devType;
630 1 : CHK_RET(hrtGetDeviceType(devType));
631 1 : if (devType != DevType::DEV_TYPE_950) {
632 0 : return HCCL_SUCCESS;
633 : }
634 :
635 1 : auto startConsistency = std::chrono::steady_clock::now();
636 1 : CHK_RET(exchangeInfoMgr_.BatchExchangeAndCheckConsistency(
637 : channelDescs, hcommDescs, channelNum, newChannels, collCommConfigConsistency_, engine));
638 0 : auto endConsistency = std::chrono::steady_clock::now();
639 : auto durationConsistency
640 0 : = std::chrono::duration_cast<std::chrono::microseconds>(endConsistency - startConsistency).count();
641 0 : HCCL_INFO(
642 : "[MyRank][%s] BatchExchangeAndCheckConsistency Time Elapsed [%lld]us, channelNum [%u]", __func__,
643 : durationConsistency, channelNum);
644 0 : return HCCL_SUCCESS;
645 : }
646 :
647 : constexpr uint32_t MEM_HANDLE_NUM_MAX = 256; // memHandleNum的默认限制最大为256
648 : constexpr uint32_t NOTIFY_NUM_MAX = 64; // notifynum 的默认限制最大为64
649 :
650 3 : HcclResult MyRank::CheckChannelParam(CommEngine engine, const HcclChannelDesc* channelDesc, uint32_t channelNum) const
651 : {
652 6 : for (u32 index = 0; index < channelNum; ++index) {
653 4 : if (engine == COMM_ENGINE_AIV) {
654 0 : CHK_PRT_RET(
655 : (channelDesc->memHandleNum > MEM_HANDLE_NUM_MAX),
656 : HCCL_ERROR(
657 : "[%s]Channeldesc[%u] invalid memHandleNum, memHandleNum[%u], max channel num[%u]", __func__, index,
658 : channelDesc->memHandleNum, MEM_HANDLE_NUM_MAX),
659 : HCCL_E_PARA);
660 0 : CHK_PRT_RET(
661 : (channelDesc->memHandleNum != 0 && channelDesc->memHandles == nullptr),
662 : HCCL_ERROR("[%s]Channeldesc[%u] invalid memHandles, memHandles is null", __func__, index), HCCL_E_PARA);
663 : } else {
664 4 : if (channelDesc->memHandleNum != 0) {
665 2 : HCCL_WARNING(
666 : "[%s]Channeldesc[%u] memHandleNum[%u] is non-zero, memHandle exchange is not supported.", __func__,
667 : index, channelDesc->memHandleNum);
668 : }
669 : }
670 4 : CHK_PRT_RET(
671 : channelDesc->notifyNum > NOTIFY_NUM_MAX,
672 : HCCL_ERROR(
673 : "[%s]Channeldesc[%u] invalid notifyNum [%u], max notify num[%u]", __func__, index,
674 : channelDesc->notifyNum, NOTIFY_NUM_MAX),
675 : HCCL_E_PARA);
676 : }
677 :
678 2 : return HCCL_SUCCESS;
679 : }
680 :
681 : // 批量创建channels，如果CCU资源不足（如Xn, Cke, channel ctx, jetty ctx, wqebb）会失败，返回HCCL_E_UNAVAIL
682 13 : HcclResult MyRank::BatchCreateChannels(
683 : CommEngine engine, const HcclChannelDesc* channelDescs, uint32_t channelNum,
684 : std::vector<HcommChannelDesc>& hcommDescs, ChannelHandle* channelHandles,
685 : std::vector<std::vector<MemHandle>>& allHandles)
686 : {
687 13 : CHK_PTR_NULL(channelDescs);
688 13 : CHK_PTR_NULL(channelHandles);
689 13 : CHK_PRT_RET(channelNum == 0, HCCL_ERROR("[%s] invalid param: channelNum is zero", __func__), HCCL_E_PARA);
690 :
691 : // 持锁保护 newChannels_/handleToEpPair_；失败路径调用的 DestroyNewChannels 由本函数持锁，内部不再加锁
692 13 : std::lock_guard<std::mutex> lock(channelIndexMtx_);
693 :
694 13 : uint32_t localRank = rankId_;
695 13 : CHK_SMART_PTR_NULL(commMems_);
696 13 : CHK_PTR_NULL(endpointMgr_);
697 : std::unordered_map<RankPair*, std::unordered_map<CommEngine, std::unordered_map<hcomm::EndpointPair*, u32>>>
698 13 : reuseChannelIdxMap{};
699 :
700 : // 记录本轮新申请的channel
701 13 : newChannels_.clear();
702 13 : bool isAllSuccess = true;
703 :
704 45 : for (uint32_t i = 0; i < channelNum; ++i) {
705 34 : const EndpointDesc& localEndpointDesc = channelDescs[i].localEndpoint;
706 34 : const EndpointDesc& remoteEndpointDesc = channelDescs[i].remoteEndpoint;
707 34 : uint32_t remoteRank = channelDescs[i].remoteRank;
708 :
709 34 : HCCL_INFO(
710 : "[%s][%u/%u] remoteRank[%u] localProtocol[%d] remoteProtocol[%d] engine[%s]", __func__, i + 1, channelNum,
711 : remoteRank, localEndpointDesc.protocol, remoteEndpointDesc.protocol,
712 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str());
713 :
714 34 : EndpointHandle epHandle = nullptr;
715 34 : auto ret = endpointMgr_->Get(localEndpointDesc, epHandle);
716 34 : CHK_PRT_RET(
717 : ret != HCCL_SUCCESS,
718 : HCCL_ERROR(
719 : "[%s] failed to get endpoint, channelIndex[%u], remoteRank[%u], protocol[%d]", __func__, i, remoteRank,
720 : localEndpointDesc.protocol),
721 : ret);
722 34 : CHK_PTR_NULL(epHandle);
723 :
724 : // 启动监听
725 34 : uint32_t listenPort = 0;
726 34 : CHK_RET(GetDevicePortInternal(localRank, &listenPort, localEndpointDesc.loc.locType));
727 34 : if (listenPort == Hccl::DEFAULT_VALUE_TCPPORT) {
728 2 : auto portRanges = Hccl::EnvConfig::GetInstance().GetHostNicConfig().GetDeviceSocketPortRange();
729 2 : if (!portRanges.empty()) {
730 1 : listenPort = portRanges[0].min;
731 1 : HCCL_INFO(
732 : "[%s] listenPort is default[%u], use port[%u] from HCCL_NPU_SOCKET_PORT_RANGE", __func__,
733 : Hccl::DEFAULT_VALUE_TCPPORT, listenPort);
734 : }
735 2 : }
736 34 : CHK_RET(static_cast<HcclResult>(HcommEndpointStartListen(epHandle, listenPort, nullptr)));
737 :
738 34 : HCCL_INFO(
739 : "[%s][%u/%u] remoteRank[%u] epHandle[%p] protocol[%d]", __func__, i + 1, channelNum, remoteRank, epHandle,
740 : localEndpointDesc.protocol);
741 :
742 : // 注册内存
743 34 : CHK_RET(PrepareMemHandles(epHandle, channelDescs[i].memHandles, channelDescs[i].memHandleNum, allHandles[i]));
744 34 : HCCL_INFO(
745 : "[%s][%u/%u] remoteRank[%u] got %zu user memory handles", __func__, i + 1, channelNum, remoteRank,
746 : allHandles[i].size());
747 :
748 34 : hcommDescs[i].exchangeAllMems = false;
749 34 : hcommDescs[i].memHandles = allHandles[i].data();
750 34 : hcommDescs[i].memHandleNum = allHandles[i].size();
751 :
752 34 : hcomm::EndpointPair* endpointPair = nullptr;
753 34 : RankIdPair rankIdPair = std::make_pair(localRank, remoteRank);
754 34 : EndpointDescPair endpointDescPair = std::make_pair(localEndpointDesc, remoteEndpointDesc);
755 34 : RankPair* rankPair = nullptr;
756 34 : CHK_RET(rankPairMgr_->Get(rankIdPair, rankPair));
757 34 : CHK_PTR_NULL(rankPair);
758 34 : CHK_RET(rankPair->GetEndpointPair(endpointDescPair, endpointPair));
759 34 : CHK_PTR_NULL(endpointPair);
760 :
761 34 : if (reuseChannelIdxMap.find(rankPair) == reuseChannelIdxMap.end()) {
762 19 : std::unordered_map<CommEngine, std::unordered_map<hcomm::EndpointPair*, u32>> engine2EndpointPairMap{};
763 19 : std::unordered_map<hcomm::EndpointPair*, u32> endpointPair2Idx{};
764 19 : endpointPair2Idx.emplace(endpointPair, 0);
765 19 : engine2EndpointPairMap.emplace(engine, endpointPair2Idx);
766 19 : reuseChannelIdxMap.emplace(rankPair, engine2EndpointPairMap);
767 34 : } else if (reuseChannelIdxMap[rankPair].find(engine) == reuseChannelIdxMap[rankPair].end()) {
768 0 : std::unordered_map<hcomm::EndpointPair*, u32> endpointPair2Idx{};
769 0 : endpointPair2Idx.emplace(endpointPair, 0);
770 0 : reuseChannelIdxMap[rankPair].emplace(engine, endpointPair2Idx);
771 0 : } else if (
772 15 : reuseChannelIdxMap[rankPair][engine].find(endpointPair) == reuseChannelIdxMap[rankPair][engine].end()) {
773 3 : reuseChannelIdxMap[rankPair][engine].emplace(endpointPair, 0);
774 : }
775 :
776 34 : u32& reuseIdx = reuseChannelIdxMap[rankPair][engine][endpointPair];
777 34 : u32 idx = reuseIdx;
778 : /* hostNIC -- DeviceNic（transport不复用link/Channel），此流程也是新创建channel，需要计入isNewChannel */
779 34 : if (localEndpointDesc.loc.locType != remoteEndpointDesc.loc.locType) {
780 0 : idx = UNREUSE_CHANNEL_IDX;
781 : }
782 34 : bool isNewChannel = (endpointPair->IsChannelNotExist(engine, reuseIdx) || (idx == UNREUSE_CHANNEL_IDX));
783 :
784 : // CreateChannel 返回 HCCL_E_UNAVAIL 表示资源不足创建失败
785 34 : ret = endpointPair->CreateChannel(epHandle, engine, idx, &hcommDescs[i], channelHandles + i);
786 34 : if (ret == HCCL_E_TIMEOUT || ret == HCCL_E_INTERNAL) {
787 0 : Hccl::TlsStatus tlsStatus = Hccl::TlsStatus::UNKNOWN;
788 0 : CHK_PRT_CONT(
789 : GetLocalTlsStatus(localEndpointDesc.loc.locType, tlsStatus) != HCCL_SUCCESS,
790 : HCCL_WARNING("[GetLocalTlsStatus] Can not get TlsStatus"));
791 0 : HCCL_ERROR(
792 : "[%s] failed to create channel, channelIndex[%u], localRank[%u], remoteRank[%u], protocol[%s], "
793 : "tlsType[%s], tlsStatus[%s], ret[%d]",
794 : __func__, i, localRank, remoteRank,
795 : MyRankUtils::GetCommProtocolEnumStr(localEndpointDesc.protocol).c_str(),
796 : MyRankUtils::GetTlsTypeStr(localEndpointDesc.loc.locType), MyRankUtils::GetTlsStatusStr(tlsStatus),
797 : ret);
798 : }
799 34 : if (ret == HCCL_E_UNAVAIL) {
800 : // 申请channel因资源不足失败，清理已申请的channel
801 2 : HCCL_RUN_WARNING(
802 : "[%s] create channel failed, channelIndex[%u], remoteRank[%u], engine[%s], reuseIdx[%u], need clean "
803 : "new channels",
804 : __func__, i + 1, remoteRank, GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), reuseIdx);
805 2 : isAllSuccess = false;
806 2 : break;
807 : }
808 : // 记录新申请的channel信息，用于清理临时资源
809 32 : if (isNewChannel) {
810 22 : newChannels_.emplace_back(std::make_pair(i, reuseIdx));
811 : }
812 :
813 32 : if (ret != HCCL_SUCCESS) {
814 0 : if (ret != HCCL_E_TIMEOUT && ret != HCCL_E_INTERNAL) {
815 0 : HCCL_ERROR(
816 : "[%s] failed to create channel, channelIndex[%u], remoteRank[%u], engine[%s], reuseIndex[%u]",
817 : __func__, i + 1, remoteRank, GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(),
818 : reuseIdx);
819 : }
820 0 : return ret;
821 : }
822 32 : if (idx != UNREUSE_CHANNEL_IDX) {
823 32 : reuseIdx++;
824 : }
825 :
826 : // 登记 handle -> EndpointPair 反查索引；真实槽位由 EndpointPair::handleToLoc_ 维护
827 32 : handleToEpPair_[channelHandles[i]] = endpointPair;
828 :
829 32 : HCCL_INFO(
830 : "[%s][%u/%u] channel created successfully, remoteRank[%u], channelHandle[%p]", __func__, i + 1, channelNum,
831 : remoteRank, channelHandles[i]);
832 : }
833 :
834 : // 如果申请失败，清理endpoint pair中记录的channel handle
835 13 : if (!isAllSuccess) {
836 2 : HCCL_RUN_WARNING(
837 : "[%s] create channel failed, destroy new channels num[%zu], engine[%s]", __func__, newChannels_.size(),
838 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str());
839 2 : CHK_RET(DestroyNewChannels(engine, channelDescs, newChannels_));
840 2 : return HCCL_E_UNAVAIL;
841 : }
842 :
843 11 : return HCCL_SUCCESS;
844 13 : }
845 :
846 3 : HcclResult MyRank::DestroyNewChannels(
847 : CommEngine engine, const HcclChannelDesc* channelDescs, const std::vector<std::pair<u32, u32>>& newChannels)
848 : {
849 3 : HcclResult firstErr = HCCL_SUCCESS;
850 3 : uint32_t localRank = rankId_;
851 12 : for (auto idxPairIter = std::rbegin(newChannels); idxPairIter != std::rend(newChannels);
852 6 : ++idxPairIter) { // 由于新申请的在申请过的后面，所以要从后往前找reuseIdx销毁
853 6 : auto idxPair = *idxPairIter;
854 6 : const EndpointDesc& localEndpointDesc = channelDescs[idxPair.first].localEndpoint;
855 6 : const EndpointDesc& remoteEndpointDesc = channelDescs[idxPair.first].remoteEndpoint;
856 6 : uint32_t remoteRank = channelDescs[idxPair.first].remoteRank;
857 6 : hcomm::EndpointPair* endpointPair = nullptr;
858 6 : RankIdPair rankIdPair = std::make_pair(localRank, remoteRank);
859 6 : EndpointDescPair endpointDescPair = std::make_pair(localEndpointDesc, remoteEndpointDesc);
860 6 : RankPair* rankPair = nullptr;
861 6 : CHK_RET(rankPairMgr_->Get(rankIdPair, rankPair));
862 6 : CHK_PTR_NULL(rankPair);
863 6 : CHK_RET(rankPair->GetEndpointPair(endpointDescPair, endpointPair));
864 6 : CHK_PTR_NULL(endpointPair);
865 : // DestroyChannel 会 erase 向量导致下标变化, 需先取出 handle
866 6 : ChannelHandle handleToErase = 0;
867 6 : endpointPair->GetChannelHandle(engine, idxPair.second, handleToErase);
868 : // 单个 channel 销毁失败不中断其余清理；记录首个错误，最终统一清空本次新建列表
869 6 : HcclResult destroyRet = endpointPair->DestroyChannel(engine, idxPair.second);
870 6 : if (destroyRet != HCCL_SUCCESS) {
871 0 : HCCL_ERROR(
872 : "[%s] DestroyChannel failed, engine[%s] reuseIdx[%u] ret[%d], continue.", __func__,
873 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), idxPair.second, destroyRet);
874 0 : if (firstErr == HCCL_SUCCESS) {
875 0 : firstErr = destroyRet;
876 : }
877 : }
878 6 : if (handleToErase != 0) {
879 2 : handleToEpPair_.erase(handleToErase);
880 : }
881 : }
882 3 : newChannels_.clear();
883 3 : return firstErr;
884 : }
885 :
886 : HcclResult
887 15 : MyRank::QueryOneChannel(CommEngine engine, const HcclChannelDesc& channelDesc, u32 reuseIdx, ChannelHandle& handle)
888 : {
889 15 : handle = 0;
890 15 : const RankIdPair rankIdPair = std::make_pair(rankId_, channelDesc.remoteRank);
891 15 : const EndpointDescPair endpointDescPair = std::make_pair(channelDesc.localEndpoint, channelDesc.remoteEndpoint);
892 :
893 15 : RankPair* rankPair = nullptr;
894 15 : if (rankPairMgr_->Find(rankIdPair, rankPair) != HCCL_SUCCESS || rankPair == nullptr) {
895 3 : return HCCL_SUCCESS;
896 : }
897 12 : hcomm::EndpointPair* epPair = nullptr;
898 12 : if (rankPair->GetEndpointPair(endpointDescPair, epPair) != HCCL_SUCCESS || epPair == nullptr) {
899 0 : return HCCL_SUCCESS;
900 : }
901 12 : ChannelHandle slotHandle = 0;
902 12 : if (epPair->GetChannelHandle(engine, reuseIdx, slotHandle)) {
903 6 : handle = slotHandle;
904 : }
905 12 : return HCCL_SUCCESS;
906 : }
907 :
908 13 : HcclResult MyRank::QueryChannels(
909 : CommEngine engine, const HcclChannelDesc* channelDescs, uint32_t channelNum, ChannelHandle* channels)
910 : {
911 13 : CHK_PTR_NULL(channelDescs);
912 12 : CHK_PTR_NULL(channels);
913 11 : CHK_PRT_RET(channelNum == 0, HCCL_ERROR("[%s] invalid param: channelNum is zero", __func__), HCCL_E_PARA);
914 :
915 10 : HCCL_INFO(
916 : "[MyRank][%s] Enter engine[%s] channelNum[%u] rankId[%u]", __func__,
917 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), channelNum, rankId_);
918 :
919 : // 与 BatchCreateChannels 保持一致的 reuseIdx 累计逻辑
920 : std::unordered_map<RankIdPair, std::unordered_map<EndpointDescPair, std::unordered_map<CommEngine, u32>>>
921 10 : reuseIdxMap{};
922 :
923 26 : for (uint32_t i = 0; i < channelNum; ++i) {
924 16 : channels[i] = 0;
925 16 : const auto& channelDesc = channelDescs[i];
926 16 : uint32_t remoteRank = channelDesc.remoteRank;
927 16 : const RankIdPair rankIdPair = std::make_pair(rankId_, remoteRank);
928 16 : const EndpointDescPair endpointDescPair = std::make_pair(channelDesc.localEndpoint, channelDesc.remoteEndpoint);
929 :
930 16 : u32& reuseIdx = reuseIdxMap[rankIdPair][endpointDescPair][engine];
931 16 : u32 idx = reuseIdx;
932 16 : if (channelDesc.localEndpoint.loc.locType != channelDesc.remoteEndpoint.loc.locType) {
933 1 : idx = UNREUSE_CHANNEL_IDX;
934 : }
935 :
936 : // 仅当非 UNREUSE 且槽位存在时返回 handle
937 16 : if (idx != UNREUSE_CHANNEL_IDX) {
938 15 : (void)QueryOneChannel(engine, channelDesc, reuseIdx, channels[i]);
939 : }
940 :
941 16 : HCCL_INFO(
942 : "[MyRank][%s] [%u/%u] remoteRank[%u] exist[%s] handle[0x%llx] reuseIdx[%u] unreuse[%d]", __func__, i + 1,
943 : channelNum, remoteRank, channels[i] != 0 ? "yes" : "no", channels[i], reuseIdx, idx == UNREUSE_CHANNEL_IDX);
944 :
945 : // 与 BatchCreateChannels 一致: 非 UNREUSE 才递增 reuseIdx(引用, 直接改 map 内值)
946 16 : if (idx != UNREUSE_CHANNEL_IDX) {
947 15 : reuseIdx++;
948 : }
949 : }
950 :
951 : // 对发生句柄转换的引擎，经平台 H2D 反向映射把 host 句柄转换为用户实际使用的句柄
952 : // （device 句柄），保证 Query 返回值与 HcclChannelAcquire 出参一致
953 10 : if (engine == COMM_ENGINE_AICPU || engine == COMM_ENGINE_AICPU_TS || engine == COMM_ENGINE_AIV) {
954 0 : for (uint32_t i = 0; i < channelNum; ++i) {
955 0 : if (channels[i] != 0) {
956 0 : ChannelHandle deviceHandle = 0;
957 0 : if (hcomm::ChannelProcess::ResolveHostHandleToDevice(channels[i], deviceHandle) == HCCL_SUCCESS
958 0 : && deviceHandle != 0) {
959 0 : channels[i] = deviceHandle;
960 : }
961 : }
962 : }
963 : }
964 10 : return HCCL_SUCCESS;
965 10 : }
966 :
967 : // 记录批量销毁过程中的首个错误与对应计数，供 DestroyOneChannel 复用
968 23 : static void RecordDestroyError(HcclResult& firstErr, u32& errCnt, HcclResult err)
969 : {
970 23 : errCnt++;
971 23 : if (firstErr == HCCL_SUCCESS) {
972 21 : firstErr = err;
973 : }
974 23 : }
975 :
976 43 : HcclResult MyRank::DestroyOneChannel(
977 : ChannelHandle userHandle, u32 index, HcclResult& firstErr, u32& invalidHandleCnt, u32& failedCnt)
978 : {
979 : // 反查索引以 host 句柄为键：AIV/AICPU_TS 入参为 device 句柄，先经 D2H 映射解析为 host 句柄
980 43 : ChannelHandle hostHandle = userHandle;
981 43 : ChannelHandle resolved = 0;
982 43 : if (hcomm::ChannelProcess::ResolveUserHandleToHost(userHandle, resolved) == HCCL_SUCCESS && resolved != 0) {
983 0 : hostHandle = resolved;
984 : }
985 43 : auto it = handleToEpPair_.find(hostHandle);
986 43 : if (it == handleToEpPair_.end()) {
987 12 : HCCL_ERROR("[%s] channel handle[0x%llx] not found, channelIndex[%u].", __func__, userHandle, index);
988 12 : RecordDestroyError(firstErr, invalidHandleCnt, HCCL_E_NOT_FOUND);
989 12 : return HCCL_SUCCESS;
990 : }
991 31 : hcomm::EndpointPair* epPair = it->second;
992 31 : if (epPair == nullptr) {
993 : // 反查索引条目为空指针（异常数据），清理并按无效句柄容错
994 1 : HCCL_ERROR("[%s] channel handle[0x%llx] endpoint pair is null, channelIndex[%u].", __func__, userHandle, index);
995 1 : handleToEpPair_.erase(it);
996 1 : RecordDestroyError(firstErr, invalidHandleCnt, HCCL_E_NOT_FOUND);
997 1 : return HCCL_SUCCESS;
998 : }
999 30 : CommEngine engine = COMM_ENGINE_RESERVED;
1000 30 : u32 reuseIdx = 0;
1001 30 : if (!epPair->FindChannelLoc(hostHandle, engine, reuseIdx)) {
1002 1 : HCCL_ERROR("[%s] channel handle[0x%llx] FindChannelLoc failed, channelIndex[%u].", __func__, userHandle, index);
1003 1 : RecordDestroyError(firstErr, invalidHandleCnt, HCCL_E_NOT_FOUND);
1004 1 : return HCCL_SUCCESS;
1005 : }
1006 : // 暂只支持 CCU 引擎： 其他场景的 channel 销毁无法保证资源完整释放
1007 29 : if (engine != COMM_ENGINE_CCU) {
1008 7 : HCCL_WARNING(
1009 : "[%s] channel handle[0x%llx] engine[%s] not supported, only CCU engine supported, channelIndex[%u].",
1010 : __func__, userHandle, GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), index);
1011 7 : RecordDestroyError(firstErr, failedCnt, HCCL_E_NOT_SUPPORT);
1012 7 : return HCCL_SUCCESS;
1013 : }
1014 22 : HcclResult destroyRet = epPair->DestroyChannel(engine, reuseIdx);
1015 22 : if (destroyRet != HCCL_SUCCESS) {
1016 2 : HCCL_ERROR("[%s] DestroyChannel failed, handle[0x%llx] ret[%d], continue.", __func__, hostHandle, destroyRet);
1017 2 : RecordDestroyError(firstErr, failedCnt, destroyRet);
1018 2 : return HCCL_SUCCESS;
1019 : }
1020 20 : handleToEpPair_.erase(it);
1021 20 : return HCCL_SUCCESS;
1022 : }
1023 :
1024 36 : HcclResult MyRank::DestroyChannels(const ChannelHandle* channels, uint32_t channelNum)
1025 : {
1026 36 : CHK_PTR_NULL(channels);
1027 35 : CHK_PRT_RET(channelNum == 0, HCCL_ERROR("[%s] invalid param: channelNum is zero", __func__), HCCL_E_PARA);
1028 :
1029 34 : std::lock_guard<std::mutex> lock(channelIndexMtx_);
1030 :
1031 34 : HCCL_INFO("[MyRank][%s] Enter channelNum[%u] rankId[%u]", __func__, channelNum, rankId_);
1032 :
1033 34 : HcclResult firstErr = HCCL_SUCCESS;
1034 34 : u32 invalidHandleCnt = 0;
1035 34 : u32 failedCnt = 0;
1036 :
1037 77 : for (uint32_t i = 0; i < channelNum; ++i) {
1038 43 : (void)DestroyOneChannel(channels[i], i, firstErr, invalidHandleCnt, failedCnt);
1039 : }
1040 :
1041 34 : if (firstErr != HCCL_SUCCESS) {
1042 21 : u32 destroyedCnt = channelNum - invalidHandleCnt - failedCnt;
1043 21 : HCCL_ERROR(
1044 : "[%s] finished with errors, total[%u] destroyed[%u] failed[%u] invalidHandle[%u] firstErr[%d].", __func__,
1045 : channelNum, destroyedCnt, failedCnt, invalidHandleCnt, static_cast<s32>(firstErr));
1046 21 : return firstErr;
1047 : }
1048 13 : return HCCL_SUCCESS;
1049 34 : }
1050 :
1051 : HcclResult
1052 2 : MyRank::BatchConnectChannels(const HcclChannelDesc* channelDescs, ChannelHandle* channelHandles, uint32_t channelNum)
1053 : {
1054 2 : auto timeout = std::chrono::seconds(Hccl::EnvConfig::GetInstance().GetSocketConfig().GetLinkTimeOut());
1055 2 : auto startTime = std::chrono::steady_clock::now();
1056 :
1057 2 : HCCL_INFO(
1058 : "[%s] start connecting channels, channelNum[%u], timeout[%lld]sec", __func__, channelNum, timeout.count());
1059 :
1060 2 : std::vector<int32_t> statusVec(channelNum, 0);
1061 2 : int32_t* statusList = statusVec.data();
1062 2 : uint32_t retryCount = 0;
1063 : while (true) {
1064 1461862 : HcclResult ret = hcomm::ChannelProcess::ChannelGetStatus(channelHandles, channelNum, statusList);
1065 :
1066 : // 卫语句：先处理异常情况
1067 :
1068 : // 1. 检查超时
1069 1461862 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
1070 : auto elapsed
1071 2 : = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - startTime)
1072 2 : .count();
1073 2 : HCCL_ERROR(
1074 : "[%s] channel connect timeout after %lld sec, channelNum[%u], elapsed[%lld]ms, retryCount[%u]",
1075 : __func__, timeout.count(), channelNum, elapsed, retryCount);
1076 14 : RPT_INPUT_ERR(
1077 : true, "EI0006", std::vector<std::string>({"reason"}),
1078 : std::vector<std::string>({GET_SOCKET_TIMEOUT_REASON_CLOSE_DETECT}));
1079 2 : std::vector<Hccl::TlsStatus> tlsStatusList(channelNum, Hccl::TlsStatus::UNKNOWN);
1080 2 : GetAbnormalChannelTlsStatus(channelDescs, statusList, channelNum, tlsStatusList);
1081 2 : logger::ChannelLogger::PrintChannelErrorDetails(
1082 : rankId_, channelNum, channelDescs, channelHandles, statusList, static_cast<uint64_t>(elapsed),
1083 2 : tlsStatusList.data());
1084 2 : return HCCL_E_TIMEOUT;
1085 2 : }
1086 :
1087 : // 2. 处理重试（去除频繁的重试日志，一秒可能重试上千次）
1088 1461860 : if (ret == HCCL_E_AGAIN) {
1089 1461860 : retryCount++;
1090 1461860 : continue;
1091 : }
1092 :
1093 : // 3. 处理资源不足（属于可预期回退场景）
1094 0 : if (ret == HCCL_E_UNAVAIL) {
1095 : auto elapsed
1096 0 : = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - startTime)
1097 0 : .count();
1098 0 : HCCL_WARNING(
1099 : "[%s] channel connect resource unavailable, channelNum[%u], elapsed[%lld]ms, retryCount[%u]", __func__,
1100 : channelNum, elapsed, retryCount);
1101 0 : return ret;
1102 : }
1103 :
1104 : // 4. 处理失败
1105 0 : if (ret != HCCL_SUCCESS) {
1106 : auto elapsed
1107 0 : = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - startTime)
1108 0 : .count();
1109 0 : HCCL_ERROR(
1110 : "[%s] channel connect failed, channelNum[%u], ret[%d], elapsed[%lld]ms, retryCount[%u]", __func__,
1111 : channelNum, ret, elapsed, retryCount);
1112 0 : std::vector<Hccl::TlsStatus> tlsStatusList(channelNum, Hccl::TlsStatus::UNKNOWN);
1113 0 : GetAbnormalChannelTlsStatus(channelDescs, statusList, channelNum, tlsStatusList);
1114 0 : logger::ChannelLogger::PrintChannelErrorDetails(
1115 : rankId_, channelNum, channelDescs, channelHandles, statusList, static_cast<uint64_t>(elapsed),
1116 0 : tlsStatusList.data());
1117 0 : return ret;
1118 0 : }
1119 :
1120 : // 5. 正常情况：所有通道连接成功
1121 : auto elapsed
1122 0 : = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - startTime)
1123 0 : .count();
1124 0 : HCCL_INFO(
1125 : "[%s] all channels connected successfully, channelNum[%u], elapsed[%lld]ms, retryCount[%u]", __func__,
1126 : channelNum, elapsed, retryCount);
1127 0 : break;
1128 1461860 : }
1129 0 : return HCCL_SUCCESS;
1130 4 : }
1131 :
1132 15 : HcclResult MyRank::ConfigSqDepthByExpansionMode(CommEngine engine, HcommChannelDesc& hcommDesc) const
1133 : {
1134 15 : const u32 configuredSqDepth = config_.GetConfigSqDepth();
1135 15 : if (configuredSqDepth != HCCL_COMM_SQ_DEPTH_CONFIG_NOT_SET) {
1136 7 : const CommProtocol remoteProtocol = hcommDesc.remoteEndpoint.protocol;
1137 7 : if (engine == COMM_ENGINE_AIV
1138 4 : && (remoteProtocol == COMM_PROTOCOL_UBC_TP || remoteProtocol == COMM_PROTOCOL_UBC_CTP
1139 2 : || remoteProtocol == COMM_PROTOCOL_UBG)) {
1140 3 : hcommDesc.ubAttr.sqDepth = configuredSqDepth;
1141 3 : return HCCL_SUCCESS;
1142 : } else {
1143 4 : HCCL_WARNING(
1144 : "[%s] configured sqDepth[%u] is not supported when engine[%s] protocol[%s].", __func__,
1145 : configuredSqDepth, GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(),
1146 : MyRankUtils::GetCommProtocolEnumStr(remoteProtocol).c_str());
1147 : }
1148 : }
1149 :
1150 12 : constexpr u32 CCU_MS_MODE_DEPTH = 128;
1151 12 : constexpr u32 CCU_SCHED_MODE_DEPTH = 16;
1152 12 : if (engine == COMM_ENGINE_CCU) {
1153 7 : if (opExpansionMode_ == CCU_MS_MODE) {
1154 2 : hcommDesc.ubAttr.sqDepth = CCU_MS_MODE_DEPTH;
1155 5 : } else if (opExpansionMode_ == CCU_SCHED_MODE) {
1156 4 : hcommDesc.ubAttr.sqDepth = CCU_SCHED_MODE_DEPTH;
1157 : } else {
1158 1 : HCCL_ERROR("[%s] unexpected op expansion mode[%u] for ccu,", __func__, opExpansionMode_);
1159 1 : return HCCL_E_INTERNAL;
1160 : }
1161 : }
1162 11 : return HCCL_SUCCESS;
1163 : }
1164 :
1165 0 : void MyRank::LogChannelCreationInfo(
1166 : CommEngine engine, const std::string& commTag, const HcclChannelDesc* channelDescs, uint32_t channelNum,
1167 : const ChannelHandle* hostChannelHandleList) const
1168 : {
1169 0 : for (u32 i = 0; i < channelNum; ++i) {
1170 0 : u32 remoteRank = channelDescs[i].remoteRank;
1171 0 : HcclCommDfx::AddChannelRemoteRankId(commTag, hostChannelHandleList[i], remoteRank);
1172 : // 打印UB通道建链信息
1173 0 : if (channelDescs[i].localEndpoint.loc.locType == ENDPOINT_LOC_TYPE_DEVICE
1174 0 : && channelDescs[i].remoteEndpoint.loc.locType == ENDPOINT_LOC_TYPE_DEVICE) {
1175 0 : HCCL_CONFIG_DEBUG(
1176 : HCCL_RES,
1177 : "create channel info:channel handle[%s] comm tag[%s] protocol[%s]"
1178 : " local rank[%u] local dev phyid[%u] remote rank[%u] remote dev phyid[%u] engine[%s]",
1179 : std::to_string(hostChannelHandleList[i]).c_str(), commTag.c_str(),
1180 : MyRankUtils::GetCommProtocolEnumStr(channelDescs[i].localEndpoint.protocol).c_str(), rankId_,
1181 : channelDescs[i].localEndpoint.loc.device.devPhyId, remoteRank,
1182 : channelDescs[i].remoteEndpoint.loc.device.devPhyId,
1183 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str());
1184 0 : } else {
1185 0 : HCCL_CONFIG_DEBUG(
1186 : HCCL_RES,
1187 : "create channel info:channel handle[%s] comm tag[%s] protocol[%s]"
1188 : " local rank[%u] remote rank[%u] engine[%s]",
1189 : std::to_string(hostChannelHandleList[i]).c_str(), commTag.c_str(),
1190 : MyRankUtils::GetCommProtocolEnumStr(channelDescs[i].localEndpoint.protocol).c_str(), rankId_,
1191 : remoteRank, GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str());
1192 : }
1193 : }
1194 0 : }
1195 :
1196 0 : HcclResult MyRank::FinalizeChannelsByEngine(
1197 : CommEngine engine, const std::string& commTag, uint32_t channelNum, std::vector<HcommChannelDesc>& hcommDescs,
1198 : ChannelHandle* hostChannelHandleList, ChannelHandle* channelHandles)
1199 : {
1200 0 : if (engine == COMM_ENGINE_AICPU || engine == COMM_ENGINE_AICPU_TS) {
1201 : // 新增：添加 kernelLaunchAicpuCommInit 调用
1202 0 : if (!callbacks_.getAicpuCommState()) {
1203 0 : HCCL_INFO("MyRank::%s kernelLaunchAicpuCommInit start.", __func__);
1204 0 : HcclResult ret = callbacks_.kernelLaunchAicpuCommInit();
1205 0 : CHK_PRT_RET(
1206 : ret != HCCL_SUCCESS, HCCL_ERROR("[%s] kernelLaunchAicpuCommInit failed, return [%d].", __func__, ret),
1207 : ret);
1208 0 : callbacks_.setAicpuCommState(true);
1209 : }
1210 0 : HcommChannelDesc* hcommDesc = hcommDescs.data();
1211 0 : CHK_RET(ChannelProcess::ChannelKernelLaunchForComm(
1212 : channelHandles, hostChannelHandleList, hcommDesc, channelNum, commTag, binHandle_));
1213 :
1214 : // ns recovery
1215 0 : nsRecoveryProcessor_->AddNsRecoveryData(engine, channelHandles, hostChannelHandleList, channelNum, commTag);
1216 :
1217 0 : return HCCL_SUCCESS;
1218 : }
1219 :
1220 0 : if (engine == COMM_ENGINE_CPU || engine == COMM_ENGINE_CCU || engine == COMM_ENGINE_AIV) {
1221 0 : CHK_SAFETY_FUNC_RET(memcpy_s(
1222 : channelHandles, channelNum * sizeof(ChannelHandle), hostChannelHandleList,
1223 : channelNum * sizeof(ChannelHandle)));
1224 0 : return HCCL_SUCCESS;
1225 : }
1226 :
1227 0 : HCCL_ERROR(
1228 : "[MyRank][%s] unsupported comm engine[%s].", __func__,
1229 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str());
1230 0 : return HCCL_E_NOT_SUPPORT;
1231 : }
1232 :
1233 6 : HcclResult MyRank::CreateChannels(
1234 : CommEngine engine, const std::string& commTag, const HcclChannelDesc* channelDescs, uint32_t channelNum,
1235 : ChannelHandle* channelHandles)
1236 : {
1237 6 : CHK_PTR_NULL(channelDescs);
1238 5 : CHK_PTR_NULL(channelHandles);
1239 4 : CHK_PRT_RET(channelNum == 0, HCCL_ERROR("[%s] invalid param: channelNum is zero", __func__), HCCL_E_PARA);
1240 :
1241 3 : HCCL_INFO(
1242 : "[CreateChannels][Enter] engine[%s] commTag[%s] channelNum[%u] rankId[%u]",
1243 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), commTag.c_str(), channelNum, rankId_);
1244 :
1245 : // 参数检查
1246 3 : CHK_RET(CheckChannelParam(engine, channelDescs, channelNum));
1247 :
1248 2 : std::vector<ChannelHandle> hostChannelHandles(channelNum);
1249 2 : ChannelHandle* hostChannelHandleList = hostChannelHandles.data();
1250 :
1251 2 : auto& rdmaConfig = Hccl::EnvConfig::GetInstance().GetRdmaConfig();
1252 4 : std::vector<HcommChannelDesc> hcommDescs(channelNum);
1253 4 : std::vector<std::vector<MemHandle>> allHandles(channelNum);
1254 : // srcPortBuffers 作为 hcommDescs[i].roceAttr.srcPortList 的底层 buffer，由 FillRoceSrcPortList 填充；
1255 : // 生命周期需覆盖到通道构造完成（HostCpuRoceChannel 构造时深拷贝到自身 srcPortBuf_）。
1256 2 : std::vector<std::vector<uint16_t>> srcPortBuffers(channelNum);
1257 5 : for (u32 i = 0; i < channelNum; ++i) {
1258 3 : hcommDescs[i] = MyRankUtils::ChannelDescHccl2Hcomm(channelDescs[i], config_);
1259 3 : hcommDescs[i].roceAttr.qpThreshold = rdmaConfig.GetRdmaMultiQpThreshold();
1260 3 : CHK_RET(ConfigSqDepthByExpansionMode(engine, hcommDescs[i]));
1261 :
1262 3 : CHK_RET(MyRankUtils::FillRoceSrcPortList(channelDescs[i], hcommDescs[i], srcPortBuffers[i]));
1263 : }
1264 :
1265 2 : auto start = std::chrono::steady_clock::now();
1266 2 : std::string socketTag = commTag + "_engine_" + std::to_string(engine);
1267 2 : CHK_RET(BatchCreateSockets(channelDescs, channelNum, socketTag, hcommDescs));
1268 2 : CHK_RET_UNAVAIL(
1269 : BatchCreateChannels(engine, channelDescs, channelNum, hcommDescs, hostChannelHandleList, allHandles));
1270 :
1271 : // 锁内快照本次新建列表：connect 阶段不再持 channelIndexMtx_，避免长耗时 IO 阻塞
1272 : // Query/Destroy；回滚时基于快照重新取锁清理，保证 newChannels_ 读写均在锁内
1273 2 : std::vector<std::pair<u32, u32>> newChannelsSnapshot;
1274 : {
1275 2 : std::lock_guard<std::mutex> lock(channelIndexMtx_);
1276 2 : newChannelsSnapshot = newChannels_;
1277 2 : }
1278 :
1279 2 : if (!newChannelsSnapshot.empty()) {
1280 1 : HcclResult connRet = BatchConnectChannels(channelDescs, hostChannelHandleList, channelNum);
1281 1 : if (connRet != HCCL_SUCCESS && engine == COMM_ENGINE_CCU) {
1282 : // CCU 场景额外回滚本次新建的 channel，避免资源残留
1283 1 : HCCL_RUN_WARNING(
1284 : "[%s] BatchConnectChannels failed[%d], engine[%s], new channels num[%u]", __func__, connRet,
1285 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), newChannelsSnapshot.size());
1286 1 : std::lock_guard<std::mutex> lock(channelIndexMtx_);
1287 1 : HcclResult destroyRet = DestroyNewChannels(engine, channelDescs, newChannelsSnapshot);
1288 1 : if (destroyRet != HCCL_SUCCESS) {
1289 0 : HCCL_ERROR(
1290 : "[%s] DestroyNewChannels failed[%d] during rollback, connRet[%d], "
1291 : "residual newChannels[%zu] may leak.",
1292 : __func__, destroyRet, connRet, newChannels_.size());
1293 : }
1294 1 : }
1295 1 : CHK_RET(connRet);
1296 0 : auto end = std::chrono::steady_clock::now();
1297 0 : auto duration = std::chrono::duration_cast<std::chrono::microseconds>(end - start).count();
1298 0 : HCCL_RUN_INFO(
1299 : "[MyRank][CreateChannels] CreateChannels Time Elapsed [%lld]us, channelNum [%u]", duration, channelNum);
1300 : }
1301 :
1302 : // 借用hcommDescs.socket，完成一致性校验必要的数据交换
1303 1 : CHK_RET(BatchExchangeAndCheckConsistency(channelDescs, hcommDescs, channelNum, newChannels_, engine));
1304 :
1305 : // 添加初始化时进行填表
1306 0 : LogChannelCreationInfo(engine, commTag, channelDescs, channelNum, hostChannelHandleList);
1307 :
1308 0 : return FinalizeChannelsByEngine(engine, commTag, channelNum, hcommDescs, hostChannelHandleList, channelHandles);
1309 2 : }
1310 :
1311 1 : HcclResult MyRank::ChannelGetHcclBuffer(ChannelHandle channel, void** buffer, uint64_t* size)
1312 : {
1313 1 : CHK_PTR_NULL(buffer);
1314 1 : CHK_PTR_NULL(size);
1315 :
1316 1 : u32 memNum = 0;
1317 1 : CommMem* remoteMem = nullptr;
1318 1 : char** memTags = nullptr;
1319 1 : CHK_RET(static_cast<HcclResult>(HcommChannelGetRemoteMems(channel, &memNum, &remoteMem, &memTags)));
1320 1 : if (memNum > 0) {
1321 0 : CHK_PTR_NULL(remoteMem);
1322 : // AicpuTsHccsChannel不使用memTag，返回为空，默认索引0为cclBuffer
1323 0 : if (memTags == nullptr) {
1324 0 : *buffer = remoteMem[0].addr;
1325 0 : *size = remoteMem[0].size;
1326 0 : HCCL_INFO("[%s] Found HcclBuffer : addr=%p, size=%llu", __func__, *buffer, *size);
1327 0 : return HCCL_SUCCESS;
1328 : }
1329 0 : for (u32 i = 0; i < memNum; ++i) {
1330 0 : std::string tag = memTags[i];
1331 0 : if (tag == "HcclBuffer") {
1332 0 : *buffer = remoteMem[i].addr;
1333 0 : *size = remoteMem[i].size;
1334 0 : HCCL_INFO("[%s] Found HcclBuffer : addr=%p, size=%llu", __func__, *buffer, *size);
1335 0 : return HCCL_SUCCESS;
1336 : }
1337 0 : HCCL_INFO("[%s] Found %s : addr=%p, size=%llu", __func__, memTags[i], remoteMem[i].addr, remoteMem[i].size);
1338 0 : }
1339 : }
1340 1 : HCCL_ERROR("[%s] HcclBuffer not found.", __func__);
1341 1 : return HCCL_E_INTERNAL;
1342 : }
1343 :
1344 : HcclResult
1345 4 : MyRank::ChannelGetRemoteMems(ChannelHandle channel, uint32_t* memNum, CommMem** remoteMem, char*** memTags) const
1346 : {
1347 4 : CHK_PTR_NULL(remoteMem);
1348 3 : CHK_PTR_NULL(memTags);
1349 2 : CHK_PTR_NULL(memNum);
1350 1 : CHK_RET(static_cast<HcclResult>(HcommChannelGetRemoteMems(channel, memNum, remoteMem, memTags)));
1351 : // 添加空指针检查，防止返回的指针为空
1352 1 : if (*memNum > 0) {
1353 0 : CHK_PTR_NULL(*remoteMem);
1354 0 : CHK_PTR_NULL(*memTags);
1355 : }
1356 1 : HCCL_INFO("[%s] success. memNum[%u]", __func__, *memNum);
1357 1 : return HCCL_SUCCESS;
1358 : }
1359 :
1360 4 : HcclResult MyRank::ChannelGetRemoteMems(
1361 : ChannelHandle channel, uint32_t* memNum, CommMem** remoteMem, std::vector<std::string>& memTags) const
1362 : {
1363 4 : CHK_PTR_NULL(remoteMem);
1364 3 : CHK_PTR_NULL(memNum);
1365 2 : char** rawTags = nullptr;
1366 2 : CHK_RET(static_cast<HcclResult>(HcommChannelGetRemoteMems(channel, memNum, remoteMem, &rawTags)));
1367 : // 添加空指针检查，防止返回的指针为空
1368 2 : if (*memNum > 0) {
1369 1 : CHK_PTR_NULL(*remoteMem);
1370 1 : CHK_PTR_NULL(rawTags);
1371 1 : memTags.reserve(*memNum);
1372 3 : for (uint32_t i = 0; i < *memNum; ++i) {
1373 2 : memTags.emplace_back(rawTags[i] == nullptr ? "" : rawTags[i]);
1374 : }
1375 : }
1376 2 : HCCL_INFO("[%s] success. memNum[%u]", __func__, *memNum);
1377 2 : return HCCL_SUCCESS;
1378 : }
1379 :
1380 0 : std::vector<ChannelHandle> MyRank::GetAllChannelList()
1381 : {
1382 0 : ChannelTable channelTable = rankPairMgr_->GetChannelTable();
1383 0 : std::vector<ChannelHandle> channelList;
1384 0 : for (const auto& rankPair : channelTable) {
1385 0 : for (const auto& endPointPair : rankPair.second) {
1386 0 : for (const auto& comEngines : endPointPair.second) {
1387 0 : channelList.insert(channelList.end(), comEngines.second.begin(), comEngines.second.end());
1388 : }
1389 : }
1390 : }
1391 :
1392 0 : return channelList;
1393 0 : }
1394 :
1395 164 : void MyRank::SetKfcControlTransfer(
1396 : std::shared_ptr<HDCommunicate> kfcControlTransferH2D, std::shared_ptr<HDCommunicate> kfcStatusTransferD2H)
1397 : {
1398 164 : if (nsRecoveryProcessor_ == nullptr) {
1399 1 : HCCL_ERROR("[MyRank][SetKfcControlTransfer] nsRecoveryProcessor_ is null, cannot set KFC control transfer.");
1400 1 : return;
1401 : }
1402 163 : nsRecoveryProcessor_->SetKfcControlTransfer(kfcControlTransferH2D, kfcStatusTransferD2H);
1403 : }
1404 :
1405 0 : HcclResult MyRank::StopLaunch()
1406 : {
1407 0 : HCCL_INFO("[NsRecovery][StopLaunch] MyRank::StopLaunch start!");
1408 0 : auto ret = nsRecoveryProcessor_->StopLaunch();
1409 0 : if (ret != HcclResult::HCCL_SUCCESS) {
1410 0 : HCCL_ERROR("[NsRecovery][StopLaunch] MyRank::StopLaunch failed, ret = 0x%016llx", HCCL_ERROR_CODE(ret));
1411 : }
1412 0 : HCCL_INFO("[NsRecovery][StopLaunch] MyRank::StopLaunch success!");
1413 0 : return ret;
1414 : }
1415 :
1416 0 : HcclResult MyRank::Clean()
1417 : {
1418 0 : HCCL_INFO("[NsRecovery][Clean] MyRank::Clean start!");
1419 0 : auto channelList = GetAllChannelList();
1420 0 : if (channelList.empty()) {
1421 0 : HCCL_INFO("[NsRecovery][Clean] Channel list empty, No need to clean!");
1422 0 : return HcclResult::HCCL_SUCCESS;
1423 : }
1424 0 : auto ret = ChannelProcess::ChannelClean(channelList.data(), channelList.size());
1425 0 : if (ret != HcclResult::HCCL_SUCCESS) {
1426 0 : HCCL_ERROR("[NsRecovery][Clean] MyRank::Clean failed, ret = 0x%016llx", HCCL_ERROR_CODE(ret));
1427 0 : return ret;
1428 : }
1429 :
1430 0 : ret = nsRecoveryProcessor_->Clean();
1431 0 : if (ret != HcclResult::HCCL_SUCCESS) {
1432 0 : HCCL_ERROR("[NsRecovery][Clean] MyRank::Clean failed, ret = 0x%016llx", HCCL_ERROR_CODE(ret));
1433 0 : return ret;
1434 : }
1435 :
1436 0 : HCCL_INFO("[NsRecovery][Clean] MyRank::Clean success!");
1437 0 : return HcclResult::HCCL_SUCCESS;
1438 0 : }
1439 :
1440 0 : HcclResult MyRank::Resume()
1441 : {
1442 0 : HCCL_INFO("[NsRecovery][Resume] MyRank::Resume start!");
1443 0 : auto channelList = GetAllChannelList();
1444 0 : if (channelList.empty()) {
1445 0 : HCCL_INFO("[NsRecovery][Resume] Resume list empty, No need to resume!");
1446 0 : return HcclResult::HCCL_SUCCESS;
1447 : }
1448 :
1449 0 : auto ret = ChannelProcess::ChannelResume(channelList.data(), channelList.size());
1450 0 : if (ret != HcclResult::HCCL_SUCCESS) {
1451 0 : HCCL_ERROR("[NsRecovery][Resume] MyRank::Resume failed, ret = 0x%016llx", HCCL_ERROR_CODE(ret));
1452 0 : return ret;
1453 : }
1454 :
1455 0 : ret = nsRecoveryProcessor_->Resume(binHandle_);
1456 0 : if (ret != HcclResult::HCCL_SUCCESS) {
1457 0 : HCCL_ERROR("[NsRecovery][Resume] MyRank::Resume failed, ret = 0x%016llx", HCCL_ERROR_CODE(ret));
1458 0 : return ret;
1459 : }
1460 :
1461 0 : HCCL_INFO("[NsRecovery][Resume] MyRank::Resume success!");
1462 0 : return HCCL_SUCCESS;
1463 0 : }
1464 :
1465 7 : CollCommConfigConsistency& MyRank::GetCollCommConfigConsistency() { return collCommConfigConsistency_; }
1466 :
1467 : } // namespace hccl
|
