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