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