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 <algorithm>
13 : #include <limits>
14 : #include <functional>
15 : #include "hccl_comm_pub.h"
16 : #include "exception_handler.h"
17 : #include "config_log.h"
18 : #include "config/env_config.h"
19 : #include "env_config/env_config_v2.h"
20 :
21 : #include "coll_comm_mgr.h"
22 : #include "hcclCommOp.h"
23 : #include "channel_process.h"
24 : #include "aicpu_ts_roce_channel_v2.h"
25 : #include "aiv_urma_channel.h"
26 : #include "hccl_group.h"
27 : #include "../resource_mgr/local/my_rank/comm_engine/kernel_launch/hccl_kernel_launch_aicpu.h"
28 : #include "param_check_basic_v2.h"
29 : #include "comm_engine_utils.h"
30 : #include "rank_consistency_checker_v2.h"
31 : #include "rank_table_crc_bridge.h"
32 : #include "hccl/hccl_types.h"
33 : #include "tp_qos.h"
34 : #include "hccl_channel_config.h"
35 : #include "shared_jetty_channel_pool.h"
36 : #include "endpoint_mgr.h"
37 : #include "hcomm_res.h"
38 : #include "channel_config.h"
39 : #include "hcclCommDfx.h"
40 : #include "coll_comm_res_c_adpt.h"
41 :
42 : using namespace hccl;
43 : /**
44 : * @note 职责：集合通信的通信域资源管理的C接口的C到C++适配
45 : */
46 :
47 : /**
48 : * @note C接口适配参考示例
49 : * @code {.c}
50 : * HcclResult HcclThreadAcquire(HcclComm comm, CommEngine engine, uint32_t threadNum,
51 : * uint32_t notifyNumPerThread, ThreadHandle *threads) {
52 : * return HCCL_SUCCESS;
53 : * }
54 : * @endcode
55 : */
56 :
57 : constexpr uint32_t HCCL_CHANNEL_VERSION_ONE = 1;
58 : constexpr uint32_t MULTIPLE = 4; // 用于A5判断TC是否为4的倍数
59 : constexpr uint32_t TC_MAX = 255; // TC的最大值（不区分芯片类型）
60 : constexpr uint32_t RETRY_INTERVAL_MIN = 5u; // retryInterval范围的最小值（不区分芯片类型）
61 : constexpr uint32_t A5_RETRY_INTERVAL_MAX = 24u; // A5的retryInterval范围的最大值
62 : constexpr uint32_t RETRY_CNT_MIN = 1u; // retryCnt范围的最小值（不区分芯片类型）
63 : constexpr uint32_t RETRY_CNT_MAX = 7u; // retryCnt范围的最大值（不区分芯片类型）
64 : constexpr uint32_t SL_MAX = 7u; // sl范围的最大值，sl即serviceLevel（不区分芯片类型）
65 : constexpr uint32_t TC_DEFAULT = 0xFFFFFFFFu; // TC的默认值（不区分芯片类型）
66 : constexpr uint32_t SL_DEFAULT = 0xFFFFFFFFu; // SL的默认值（不区分芯片类型）
67 : constexpr uint32_t kDscpToRoceTcShift = 2U; // RoCE TC = DSCP << 2（DiffServ 高 6 位为 DSCP）
68 :
69 3 : static uint32_t ResolveRoceDevPhyId(const HcclChannelDesc& channelDesc)
70 : {
71 3 : if (channelDesc.localEndpoint.loc.locType == ENDPOINT_LOC_TYPE_DEVICE) {
72 3 : return channelDesc.localEndpoint.loc.device.devPhyId;
73 : }
74 0 : s32 deviceLogicId = 0;
75 0 : u32 devicePhyId = 0U;
76 0 : if (hrtGetDevice(&deviceLogicId) != HCCL_SUCCESS) {
77 0 : return 0U;
78 : }
79 0 : if (hrtGetDevicePhyIdByIndex(static_cast<u32>(deviceLogicId), devicePhyId) != HCCL_SUCCESS) {
80 0 : return 0U;
81 : }
82 0 : return devicePhyId;
83 : }
84 :
85 8 : static void FillRoceQos(
86 : const hccl::CommConfig& commConfig, const Hccl::EnvRdmaConfig& rdmaConfig, const HcclChannelDesc& channelDesc,
87 : uint8_t& slOut, uint8_t& tcOut)
88 : {
89 8 : const uint32_t hcclQos = commConfig.GetConfigHcclQos();
90 8 : if (hcclQos == HCCL_COMM_QOS_CONFIG_NOT_SET) {
91 15 : tcOut = static_cast<uint8_t>(
92 5 : (commConfig.GetConfigTrafficClass() == INVALID_UINT) ? rdmaConfig.GetRdmaTrafficClass() :
93 0 : commConfig.GetConfigTrafficClass());
94 15 : slOut = static_cast<uint8_t>(
95 5 : (commConfig.GetConfigServiceLevel() == INVALID_UINT) ? rdmaConfig.GetRdmaServerLevel() :
96 0 : commConfig.GetConfigServiceLevel());
97 5 : return;
98 : }
99 :
100 3 : slOut = static_cast<uint8_t>(hcclQos & 0xFFU);
101 3 : const uint32_t devPhyId = ResolveRoceDevPhyId(channelDesc);
102 3 : uint8_t dscp = Hccl::kUboeDefaultDscp;
103 3 : (void)Hccl::TpQosGetDscpByQosFromHccnCfg(devPhyId, slOut, dscp);
104 3 : tcOut = static_cast<uint8_t>((static_cast<uint32_t>(dscp) << kDscpToRoceTcShift) & 0xFFU);
105 3 : HCCL_INFO(
106 : "[FillRoceQos] hcclQos compat: hcclQos[%u] devPhyId[%u] dscp[%u] sl[%u] tc[%u].", hcclQos, devPhyId,
107 : static_cast<unsigned>(dscp), static_cast<unsigned>(slOut), static_cast<unsigned>(tcOut));
108 : }
109 :
110 8 : static u32 ResolveQueueNum(const Hccl::EnvRdmaConfig& rdmaConfig, const HcclChannelDesc& channelDesc)
111 : {
112 8 : if (channelDesc.roceAttr.queueNum != INVALID_UINT) { // 用户有配置qp数量，使用用户配置的
113 8 : return channelDesc.roceAttr.queueNum;
114 : }
115 : // 查询channelDesc，localEndpoint与remoteEndpoint的CommAddr字段，得到ip对
116 0 : const auto& qpSrcPortConfig = rdmaConfig.GetMultiQpSrcPortConfig();
117 0 : const CommAddr& localCommAddr = channelDesc.localEndpoint.commAddr;
118 0 : const CommAddr& remoteCommAddr = channelDesc.remoteEndpoint.commAddr;
119 0 : char localIpStr[INET6_ADDRSTRLEN] = {0};
120 0 : char remoteIpStr[INET6_ADDRSTRLEN] = {0};
121 0 : s32 localFamily = (localCommAddr.type == COMM_ADDR_TYPE_IP_V6) ? AF_INET6 : AF_INET;
122 0 : s32 remoteFamily = (remoteCommAddr.type == COMM_ADDR_TYPE_IP_V6) ? AF_INET6 : AF_INET;
123 0 : const void* localSrc = (localFamily == AF_INET6) ? static_cast<const void*>(&localCommAddr.addr6) :
124 : static_cast<const void*>(&localCommAddr.addr);
125 0 : const void* remoteSrc = (remoteFamily == AF_INET6) ? static_cast<const void*>(&remoteCommAddr.addr6) :
126 : static_cast<const void*>(&remoteCommAddr.addr);
127 0 : (void)inet_ntop(localFamily, localSrc, localIpStr, sizeof(localIpStr));
128 0 : (void)inet_ntop(remoteFamily, remoteSrc, remoteIpStr, sizeof(remoteIpStr));
129 0 : Hccl::IpAddress localIp(localIpStr, localFamily);
130 0 : Hccl::IpAddress remoteIp(remoteIpStr, remoteFamily);
131 : // 根据ip对，查HCCL_RDMA_QP_PORT_CONFIG_PATH环境变量对应的源端口号
132 0 : u32 srcPortNum = Hccl::GetMultiQpPortsNumByIpPair(qpSrcPortConfig, localIp, remoteIp);
133 0 : if (srcPortNum > 0) { // 查看源端口号是否有配置，有则使用
134 0 : return srcPortNum;
135 : }
136 0 : return rdmaConfig.GetRdmaQueueNum();
137 : }
138 :
139 8 : static void FillChannelDescFinal(
140 : hccl::CommConfig commConfig, const HcclChannelDesc& channelDesc, HcclChannelDesc& channelDescFinal,
141 : bool isCommunicatorV2)
142 : {
143 8 : if (isCommunicatorV2) { // A5
144 8 : auto& rdmaConfig = Hccl::EnvConfig::GetInstance().GetRdmaConfig();
145 8 : channelDescFinal.roceAttr.retryCnt = (channelDesc.roceAttr.retryCnt == INVALID_UINT) ?
146 0 : rdmaConfig.GetRdmaRetryCnt() :
147 : channelDesc.roceAttr.retryCnt;
148 8 : channelDescFinal.roceAttr.retryInterval = (channelDesc.roceAttr.retryInterval == INVALID_UINT) ?
149 0 : rdmaConfig.GetRdmaTimeOut() :
150 : channelDesc.roceAttr.retryInterval;
151 8 : FillRoceQos(commConfig, rdmaConfig, channelDesc, channelDescFinal.roceAttr.sl, channelDescFinal.roceAttr.tc);
152 8 : channelDescFinal.roceAttr.queueNum = ResolveQueueNum(rdmaConfig, channelDesc);
153 : } else {
154 0 : channelDescFinal.roceAttr.retryCnt = (channelDesc.roceAttr.retryCnt == INVALID_UINT) ?
155 0 : EnvConfig::GetExternalInputRdmaRetryCnt() :
156 : channelDesc.roceAttr.retryCnt;
157 0 : channelDescFinal.roceAttr.retryInterval = (channelDesc.roceAttr.retryInterval == INVALID_UINT) ?
158 0 : EnvConfig::GetExternalInputRdmaTimeOut() :
159 : channelDesc.roceAttr.retryInterval;
160 0 : channelDescFinal.roceAttr.tc = (channelDesc.roceAttr.tc == 0xFF) ?
161 0 : EnvConfig::GetExternalInputRdmaTrafficClass() :
162 : channelDesc.roceAttr.tc;
163 0 : channelDescFinal.roceAttr.sl = (channelDesc.roceAttr.sl == 0xFF) ?
164 0 : EnvConfig::GetExternalInputRdmaServerLevel() :
165 : channelDesc.roceAttr.sl;
166 0 : channelDescFinal.roceAttr.queueNum = (channelDesc.roceAttr.queueNum == INVALID_UINT) ?
167 0 : GetExternalInputQpsPerConnection() :
168 : channelDesc.roceAttr.queueNum;
169 : }
170 8 : }
171 :
172 12 : static HcclResult CheckA5Config(hccl::CommConfig commConfig, const HcclChannelDesc& channelDesc)
173 : {
174 12 : u32 tc = commConfig.GetConfigTrafficClass();
175 12 : CHK_PRT_RET(
176 : (tc != TC_DEFAULT) && (tc > TC_MAX || (tc % MULTIPLE != 0)),
177 : HCCL_ERROR(
178 : "[ProcessRoceChannelDesc]errNo[0x%016llx] invalid hcclRdmaTrafficClass[%u], must be 0xFFFFFFFF or in "
179 : "[0,255] and a multiple of 4",
180 : static_cast<unsigned long long>(HCCL_ERROR_CODE(HCCL_E_PARA)), tc),
181 : HCCL_E_PARA);
182 :
183 11 : u32 sl = commConfig.GetConfigServiceLevel();
184 11 : CHK_PRT_RET(
185 : (sl != SL_DEFAULT) && (sl > SL_MAX),
186 : HCCL_ERROR(
187 : "[ProcessRoceChannelDesc]errNo[0x%016llx] invalid hcclRdmaServiceLevel[%u], must be 0xFFFFFFFF or in [0,7]",
188 : static_cast<unsigned long long>(HCCL_ERROR_CODE(HCCL_E_PARA)), sl),
189 : HCCL_E_PARA);
190 :
191 10 : u32 retryInterval = channelDesc.roceAttr.retryInterval;
192 10 : CHK_PRT_RET(
193 : (retryInterval != INVALID_UINT)
194 : && (retryInterval < RETRY_INTERVAL_MIN || retryInterval > A5_RETRY_INTERVAL_MAX),
195 : HCCL_ERROR(
196 : "[ProcessRoceChannelDesc]errNo[0x%016llx] invalid hcclRdmaRetryInterval[%u], must be 0xFFFFFFFF or in "
197 : "[5,24]",
198 : static_cast<unsigned long long>(HCCL_ERROR_CODE(HCCL_E_PARA)), retryInterval),
199 : HCCL_E_PARA);
200 :
201 9 : u32 retryCnt = channelDesc.roceAttr.retryCnt;
202 9 : CHK_PRT_RET(
203 : (retryCnt != INVALID_UINT) && (retryCnt < RETRY_CNT_MIN || retryCnt > RETRY_CNT_MAX),
204 : HCCL_ERROR(
205 : "[ProcessRoceChannelDesc]errNo[0x%016llx] invalid hcclRdmaRetryCnt[%u], must be 0xFFFFFFFF or in [1,7]",
206 : static_cast<unsigned long long>(HCCL_ERROR_CODE(HCCL_E_PARA)), retryCnt),
207 : HCCL_E_PARA);
208 8 : return HCCL_SUCCESS;
209 : }
210 :
211 : HcclResult
212 12 : ProcessRoceChannelDesc(const HcclChannelDesc& channelDesc, HcclChannelDesc& channelDescFinal, hccl::hcclComm* hcclComm)
213 : {
214 12 : bool isCommunicatorV2 = hcclComm->IsCommunicatorV2();
215 12 : hccl::CommConfig commConfig{}; // A5使用
216 12 : if (isCommunicatorV2) { // A5
217 12 : hccl::CollComm* collComm = hcclComm->GetCollComm();
218 12 : CHK_PTR_NULL(collComm);
219 12 : commConfig = collComm->GetCommConfig();
220 12 : CHK_RET(CheckA5Config(commConfig, channelDesc));
221 : }
222 8 : FillChannelDescFinal(commConfig, channelDesc, channelDescFinal, isCommunicatorV2);
223 8 : HCCL_INFO(
224 : "[%s]queueNum[%u], retryCnt[%u], retryInterval[%u], tc[%u], sl[%u]", __func__,
225 : channelDescFinal.roceAttr.queueNum, channelDescFinal.roceAttr.retryCnt, channelDescFinal.roceAttr.retryInterval,
226 : channelDescFinal.roceAttr.tc, channelDescFinal.roceAttr.sl);
227 8 : return HCCL_SUCCESS;
228 12 : }
229 :
230 9 : HcclResult ProcessUbChannelDesc(
231 : const HcclChannelDesc& channelDesc, const HcclChannelDesc& channelDescFinal, const hccl::hcclComm* hcclComm)
232 : {
233 : (void)channelDescFinal;
234 : (void)hcclComm;
235 :
236 9 : if (channelDesc.channelProtocol != COMM_PROTOCOL_UB_CTP && channelDesc.channelProtocol != COMM_PROTOCOL_UBC_TP
237 6 : && channelDesc.channelProtocol != COMM_PROTOCOL_UBOE && channelDesc.channelProtocol != COMM_PROTOCOL_UB_RTP) {
238 2 : HCCL_ERROR(
239 : "[%s] unexpected channelProtocol[%d], expect UB_CTP/UBC_TP/UBOE/UB_RTP", __func__,
240 : static_cast<int>(channelDesc.channelProtocol));
241 2 : return HCCL_E_PARA;
242 : }
243 7 : HCCL_INFO(
244 : "[%s] channelProtocol[%d] ub comm-domain qos applied in HcommChannelDesc::qos when converting (HcclChannelDesc "
245 : "has no qos field)",
246 : __func__, static_cast<int>(channelDesc.channelProtocol));
247 7 : return HCCL_SUCCESS;
248 : }
249 :
250 : HcclResult
251 19 : ProcessHcclChannelDesc(const HcclChannelDesc& channelDesc, HcclChannelDesc& channelDescFinal, hccl::hcclComm* hcclComm)
252 : {
253 19 : channelDescFinal.remoteRank = channelDesc.remoteRank;
254 19 : channelDescFinal.channelProtocol = channelDesc.channelProtocol;
255 19 : channelDescFinal.localEndpoint = channelDesc.localEndpoint;
256 19 : channelDescFinal.remoteEndpoint = channelDesc.remoteEndpoint;
257 19 : channelDescFinal.notifyNum = channelDesc.notifyNum;
258 19 : channelDescFinal.memHandles = channelDesc.memHandles;
259 19 : channelDescFinal.memHandleNum = channelDesc.memHandleNum;
260 :
261 : // 根据协议类型拷贝union中的相应成员
262 19 : switch (channelDesc.channelProtocol) {
263 2 : case COMM_PROTOCOL_HCCS:
264 : case COMM_PROTOCOL_HCCS_ONLY:
265 : case COMM_PROTOCOL_PCIE:
266 : case COMM_PROTOCOL_SIO:
267 2 : break;
268 2 : case COMM_PROTOCOL_UB_MEM:
269 2 : channelDescFinal.ubMemAttr.pathMode = channelDesc.ubMemAttr.pathMode;
270 2 : HCCL_INFO("[%s] ubMemAttr.pathMode[%u]", __func__, channelDescFinal.ubMemAttr.pathMode);
271 2 : break;
272 3 : case COMM_PROTOCOL_UB_CTP:
273 : case COMM_PROTOCOL_UBC_TP:
274 : case COMM_PROTOCOL_UBOE:
275 : case COMM_PROTOCOL_UB_RTP:
276 3 : return ProcessUbChannelDesc(channelDesc, channelDescFinal, hcclComm);
277 12 : case COMM_PROTOCOL_ROCE:
278 12 : return ProcessRoceChannelDesc(channelDesc, channelDescFinal, hcclComm);
279 0 : default: {
280 0 : auto ProtocolToString = [](const CommProtocol proto) -> const char* {
281 0 : switch (proto) {
282 0 : case COMM_PROTOCOL_HCCS:
283 0 : return "COMM_PROTOCOL_HCCS";
284 0 : case COMM_PROTOCOL_PCIE:
285 0 : return "COMM_PROTOCOL_PCIE";
286 0 : case COMM_PROTOCOL_SIO:
287 0 : return "COMM_PROTOCOL_SIO";
288 0 : case COMM_PROTOCOL_UB_CTP:
289 0 : return "COMM_PROTOCOL_UB_CTP";
290 0 : case COMM_PROTOCOL_UB_MEM:
291 0 : return "COMM_PROTOCOL_UB_MEM";
292 0 : case COMM_PROTOCOL_ROCE:
293 0 : return "COMM_PROTOCOL_ROCE";
294 0 : case COMM_PROTOCOL_UBC_TP:
295 0 : return "COMM_PROTOCOL_UBC_TP";
296 0 : case COMM_PROTOCOL_UBOE:
297 0 : return "COMM_PROTOCOL_UBOE";
298 0 : case COMM_PROTOCOL_UB_RTP:
299 0 : return "COMM_PROTOCOL_UB_RTP";
300 0 : case COMM_PROTOCOL_HCCS_ONLY:
301 0 : return "COMM_PROTOCOL_HCCS_ONLY";
302 0 : default:
303 0 : return "UNKNOWN_PROTOCOL";
304 : }
305 : };
306 0 : HCCL_ERROR(
307 : "[%s] Unsupported protocol[%s] found in HcclChannelDesc.", __func__,
308 : ProtocolToString(channelDesc.channelProtocol));
309 0 : return HCCL_E_PARA;
310 : }
311 : }
312 4 : return HCCL_SUCCESS;
313 : }
314 :
315 : HcclResult
316 16 : ProcessHcclResPackReq(const HcclChannelDesc& channelDesc, HcclChannelDesc& channelDescFinal, hccl::hcclComm* hcclComm)
317 : {
318 16 : if (channelDesc.header.size < channelDescFinal.header.size) {
319 : // 需要前向兼容HcclChannelDesc，末尾部分字段不支持处理
320 16 : } else if (channelDesc.header.size > channelDescFinal.header.size) {
321 : // 需要后向向兼容HcclChannelDesc，末尾部分字段会被忽略
322 : }
323 :
324 16 : if (channelDesc.header.magicWord != channelDescFinal.header.magicWord) {
325 0 : HCCL_ERROR(
326 : "[%s]channelDescFinal.header.magicWord[%u] not equal to channelDesc.header.magicWord[%u]", __func__,
327 : channelDescFinal.header.magicWord, channelDesc.header.magicWord);
328 0 : return HCCL_E_PARA;
329 : }
330 :
331 16 : uint32_t copySize = (channelDescFinal.header.size < channelDesc.header.size ? channelDescFinal.header.size :
332 16 : channelDesc.header.size)
333 0 : - sizeof(CommAbiHeader);
334 16 : CHK_SAFETY_FUNC_RET(memcpy_s(
335 : reinterpret_cast<uint8_t*>(&channelDescFinal) + sizeof(CommAbiHeader), copySize,
336 : reinterpret_cast<const uint8_t*>(&channelDesc) + sizeof(CommAbiHeader), copySize));
337 :
338 16 : if (channelDesc.header.version >= HCCL_CHANNEL_VERSION_ONE) {
339 16 : CHK_RET(ProcessHcclChannelDesc(channelDesc, channelDescFinal, hcclComm));
340 : }
341 :
342 12 : if (channelDesc.header.version > HCCL_CHANNEL_VERSION) {
343 : // 传入的版本高于当前版本，警告不支持的配置项将被忽略
344 0 : HCCL_WARNING(
345 : "The version of provided [%u] is higher than the current version[%u], "
346 : "unsupported configuration will be ignored.",
347 : channelDesc.header.version, HCCL_CHANNEL_VERSION);
348 12 : } else if (channelDesc.header.version < HCCL_CHANNEL_VERSION) {
349 : // 传入的版本低于当前版本，警告高版本支持的配置项将被忽略
350 0 : HCCL_WARNING(
351 : "The version of provided [%u] is lower than the current version[%u], "
352 : "configurations supported by later versions will be ignored.",
353 : channelDesc.header.version, HCCL_CHANNEL_VERSION);
354 : }
355 :
356 : // 如果扩展到version=2后
357 : // 1) 在底层为新的结构体和版本（version为2）上，会正常执行下面的判断处理逻辑；
358 : // 2) 在底层为旧的结构体和版本（version为1）上，下面的逻辑没有，version的2 > 1的部分会被忽略掉；
359 12 : if (channelDesc.header.version >= 2) {
360 : }
361 :
362 12 : return HCCL_SUCCESS;
363 : }
364 :
365 : static HcclResult
366 1 : BuildAivDeviceChannelEntity(const HcclChannelDesc& channelDesc, ChannelHandle hostChannel, ChannelHandle& deviceChannel)
367 : {
368 1 : void* channel = nullptr;
369 1 : CHK_RET(hcomm::ChannelProcess::ChannelGet(hostChannel, &channel));
370 1 : hcomm::Channel* baseChannel = static_cast<hcomm::Channel*>(channel);
371 1 : CHK_PTR_NULL(baseChannel);
372 :
373 1 : if (channelDesc.channelProtocol == COMM_PROTOCOL_ROCE) {
374 0 : auto* aicpuTsRoceChannelV2 = dynamic_cast<hcomm::AicpuTsRoceChannelV2*>(baseChannel);
375 0 : CHK_PTR_NULL(aicpuTsRoceChannelV2);
376 0 : HCCL_INFO(
377 : "[%s] build AIV direct device channel by AICPU+Host RoCE flow, protocol[%d], "
378 : "hostHandle[0x%llx]",
379 : __func__, channelDesc.channelProtocol, static_cast<unsigned long long>(hostChannel));
380 0 : CHK_RET(aicpuTsRoceChannelV2->BuildAndGetDevChannelEntity(&deviceChannel));
381 0 : return HCCL_SUCCESS;
382 : }
383 :
384 1 : if (channelDesc.channelProtocol == COMM_PROTOCOL_UB_CTP || channelDesc.channelProtocol == COMM_PROTOCOL_UBC_TP
385 1 : || channelDesc.channelProtocol == COMM_PROTOCOL_UB_RTP) {
386 1 : auto* aivUrmaChannel = dynamic_cast<hcomm::AivUrmaChannel*>(baseChannel);
387 1 : CHK_PTR_NULL(aivUrmaChannel);
388 1 : HCCL_INFO(
389 : "[%s] build AIV direct device channel by AIV+URMA flow, protocol[%d], "
390 : "hostHandle[0x%llx]",
391 : __func__, channelDesc.channelProtocol, static_cast<unsigned long long>(hostChannel));
392 1 : void* devChannelEntity = nullptr;
393 1 : CHK_RET(aivUrmaChannel->BuildChannelEntityToDevice(&devChannelEntity));
394 1 : CHK_PTR_NULL(devChannelEntity);
395 1 : deviceChannel = static_cast<ChannelHandle>(reinterpret_cast<uintptr_t>(devChannelEntity));
396 1 : return HCCL_SUCCESS;
397 : }
398 :
399 0 : HCCL_ERROR("[%s] protocol[%d] is not AIV direct channel protocol", __func__, channelDesc.channelProtocol);
400 0 : return HCCL_E_PARA;
401 : }
402 :
403 4 : static HcclResult ConvertAivChannelHandlesToDevicePtrs(
404 : CommEngine engine, const HcclChannelDesc* channelDescs, uint32_t channelNum, ChannelHandle* channels)
405 : {
406 4 : if (engine != COMM_ENGINE_AIV) {
407 3 : return HCCL_SUCCESS;
408 : }
409 :
410 1 : std::vector<ChannelHandle> hostChannels(channels, channels + channelNum);
411 1 : std::vector<ChannelHandle> deviceChannels(hostChannels);
412 1 : std::vector<ChannelHandle> mappedDeviceChannels;
413 1 : std::vector<ChannelHandle> mappedHostChannels;
414 2 : for (uint32_t idx = 0; idx < channelNum; ++idx) {
415 1 : if (channelDescs[idx].channelProtocol != COMM_PROTOCOL_ROCE
416 1 : && channelDescs[idx].channelProtocol != COMM_PROTOCOL_UB_CTP
417 1 : && channelDescs[idx].channelProtocol != COMM_PROTOCOL_UBC_TP
418 1 : && channelDescs[idx].channelProtocol != COMM_PROTOCOL_UB_RTP) {
419 0 : continue;
420 : }
421 1 : CHK_RET(BuildAivDeviceChannelEntity(channelDescs[idx], hostChannels[idx], deviceChannels[idx]));
422 1 : mappedDeviceChannels.emplace_back(deviceChannels[idx]);
423 1 : mappedHostChannels.emplace_back(hostChannels[idx]);
424 1 : HCCL_INFO(
425 : "[%s] convert AIV channel success, idx[%u], protocol[%d], hostHandle[0x%llx], devEntity[0x%llx]", __func__,
426 : idx, channelDescs[idx].channelProtocol, static_cast<unsigned long long>(hostChannels[idx]),
427 : static_cast<unsigned long long>(deviceChannels[idx]));
428 : }
429 :
430 1 : if (!mappedDeviceChannels.empty()) {
431 1 : CHK_RET(hcomm::ChannelProcess::RegisterChannelD2HMap(
432 : mappedDeviceChannels.data(), mappedHostChannels.data(),
433 : static_cast<uint32_t>(mappedDeviceChannels.size())));
434 : }
435 :
436 2 : for (uint32_t idx = 0; idx < channelNum; ++idx) {
437 1 : channels[idx] = deviceChannels[idx];
438 : }
439 1 : return HCCL_SUCCESS;
440 1 : }
441 2 : static bool IsUbUrmaChannelProtocol(CommProtocol protocol)
442 : {
443 2 : return protocol == COMM_PROTOCOL_UB_CTP || protocol == COMM_PROTOCOL_UBC_TP || protocol == COMM_PROTOCOL_UBOE
444 4 : || protocol == COMM_PROTOCOL_UB_RTP;
445 : }
446 :
447 2 : static bool HasUbUrmaChannel(const std::vector<HcclChannelDesc>& channelDescFinals)
448 : {
449 3 : for (const HcclChannelDesc& channelDesc : channelDescFinals) {
450 2 : if (IsUbUrmaChannelProtocol(channelDesc.channelProtocol)) {
451 1 : return true;
452 : }
453 : }
454 1 : return false;
455 : }
456 :
457 0 : static void AppendUniqueMemHandle(std::vector<HcclMemHandle>& mergedHandles, HcclMemHandle memHandle)
458 : {
459 0 : if (memHandle == nullptr) {
460 0 : return;
461 : }
462 0 : if (std::find(mergedHandles.begin(), mergedHandles.end(), memHandle) == mergedHandles.end()) {
463 0 : mergedHandles.emplace_back(memHandle);
464 : }
465 : }
466 :
467 0 : static HcclResult MergeSymmetricMemHandles(
468 : HcclChannelDesc& channelDesc, const std::vector<HcclMemHandle>& symmetricMemHandles,
469 : std::vector<HcclMemHandle>& mergedHandles)
470 : {
471 0 : if (!IsUbUrmaChannelProtocol(channelDesc.channelProtocol)) {
472 0 : return HCCL_SUCCESS;
473 : }
474 0 : if (channelDesc.memHandleNum != 0) {
475 0 : CHK_PTR_NULL(channelDesc.memHandles);
476 0 : for (uint32_t handleIdx = 0; handleIdx < channelDesc.memHandleNum; ++handleIdx) {
477 0 : AppendUniqueMemHandle(mergedHandles, channelDesc.memHandles[handleIdx]);
478 : }
479 : }
480 0 : for (HcclMemHandle memHandle : symmetricMemHandles) {
481 0 : AppendUniqueMemHandle(mergedHandles, memHandle);
482 : }
483 0 : CHK_PRT_RET(
484 : mergedHandles.size() > static_cast<size_t>(std::numeric_limits<uint32_t>::max()),
485 : HCCL_ERROR("[MergeSymmetricMemHandles] merged memHandleNum[%zu] exceeds uint32 max.", mergedHandles.size()),
486 : HCCL_E_PARA);
487 0 : channelDesc.memHandles = mergedHandles.data();
488 0 : channelDesc.memHandleNum = static_cast<uint32_t>(mergedHandles.size());
489 0 : return HCCL_SUCCESS;
490 : }
491 :
492 2 : static HcclResult AppendSymmetricMemHandles(
493 : hccl::CollComm* collComm, std::vector<HcclChannelDesc>& channelDescFinals,
494 : std::vector<std::vector<HcclMemHandle>>& mergedMemHandles, bool& hasSymmetricMemHandles)
495 : {
496 2 : CHK_PTR_NULL(collComm);
497 2 : hasSymmetricMemHandles = false;
498 2 : if (!HasUbUrmaChannel(channelDescFinals)) {
499 1 : return HCCL_SUCCESS;
500 : }
501 : // 只有UB/URMA类channel需要追加symmetric memHandle参与建链交换。
502 1 : std::vector<HcclMemHandle> symmetricMemHandles;
503 1 : CHK_RET(collComm->RegisterPendingSymmetricMemHandles(symmetricMemHandles));
504 1 : if (symmetricMemHandles.empty()) {
505 1 : return HCCL_SUCCESS;
506 : }
507 0 : hasSymmetricMemHandles = true;
508 :
509 0 : mergedMemHandles.clear();
510 0 : mergedMemHandles.resize(channelDescFinals.size());
511 0 : for (size_t idx = 0; idx < channelDescFinals.size(); ++idx) {
512 0 : CHK_RET(MergeSymmetricMemHandles(channelDescFinals[idx], symmetricMemHandles, mergedMemHandles[idx]));
513 : }
514 0 : HCCL_INFO(
515 : "[AppendSymmetricMemHandles] append symmetric memHandles success, channelNum[%zu], symMemHandleNum[%zu], "
516 : "protocols[UB_CTP/UBC_TP/UBOE].",
517 : channelDescFinals.size(), symmetricMemHandles.size());
518 0 : return HCCL_SUCCESS;
519 1 : }
520 :
521 0 : static HcclResult UpdateSymmetricRemoteMems(
522 : hccl::CollComm* collComm, const hccl::MyRank* myRank, const std::vector<HcclChannelDesc>& channelDescFinals,
523 : const ChannelHandle* channels, uint32_t channelNum)
524 : {
525 0 : CHK_PTR_NULL(collComm);
526 0 : CHK_PTR_NULL(myRank);
527 0 : CHK_PTR_NULL(channels);
528 0 : for (uint32_t idx = 0; idx < channelNum; ++idx) {
529 0 : const HcclChannelDesc& channelDesc = channelDescFinals[idx];
530 0 : if (!IsUbUrmaChannelProtocol(channelDesc.channelProtocol)) {
531 0 : continue;
532 : }
533 0 : CommMem* remoteMems = nullptr;
534 0 : uint32_t memNum = 0;
535 0 : std::vector<std::string> memTags;
536 : // CreateChannels完成后，从channel取回交换到的remoteMem/memTag并回填window。
537 0 : CHK_RET(myRank->ChannelGetRemoteMems(channels[idx], &memNum, &remoteMems, memTags));
538 0 : if (memNum == 0) {
539 0 : continue;
540 : }
541 0 : CHK_RET(collComm->UpdateSymmetricRemoteMem(channelDesc.remoteRank, remoteMems, memTags));
542 0 : }
543 0 : return HCCL_SUCCESS;
544 : }
545 :
546 7 : bool CheckCommEngine(const CommEngine engine, const uint32_t opExpansionMode)
547 : {
548 7 : constexpr uint32_t DEFAULT_MODE = 0;
549 7 : constexpr uint32_t CCU_MS_MODE = 5;
550 7 : constexpr uint32_t CCU_SCHE_MODE = 6;
551 7 : if (engine == CommEngine::COMM_ENGINE_CCU) {
552 0 : return opExpansionMode == DEFAULT_MODE || opExpansionMode == CCU_MS_MODE || opExpansionMode == CCU_SCHE_MODE;
553 : }
554 :
555 7 : return true;
556 : }
557 :
558 9 : static bool IsAicpuEngine(CommEngine engine) { return engine == COMM_ENGINE_AICPU || engine == COMM_ENGINE_AICPU_TS; }
559 :
560 : constexpr uint32_t CHANNEL_NUM_MAX = 1024 * 1024; // channel的默认限制最大为1024 * 1024
561 :
562 5 : HcclResult RegisterToClusterMonitor(HcclComm comm)
563 : {
564 5 : HCCL_INFO("[%s] START, comm[%p].", __func__, comm);
565 5 : CHK_PRT_RET(comm == nullptr, HCCL_ERROR("[%s] comm is null", __func__), HCCL_E_PTR);
566 5 : auto* hcclComm = static_cast<hccl::hcclComm*>(comm);
567 5 : CHK_PTR_NULL(hcclComm);
568 5 : if (!hcclComm->IsCommunicatorV2()) {
569 0 : HCCL_ERROR("[%s] comm is not support", __func__);
570 0 : return HCCL_E_NOT_SUPPORT;
571 : }
572 5 : hccl::CollComm* collComm = hcclComm->GetCollComm();
573 5 : CHK_PTR_NULL(collComm);
574 5 : CHK_RET(CollCommMgr::GetInstance().GetClusterMonitor(collComm->GetDeviceLogicId()).RegisterToClusterMonitor(comm));
575 3 : HCCL_INFO("%s Success", __func__);
576 3 : return HCCL_SUCCESS;
577 : }
578 :
579 : // V2 通信域 channel acquire 公共前置准备：一致性记录、引擎校验、debug 初始化、集群监控注册。
580 : // 非共享路径 HcclChannelAcquire 与共享路径 HcclChannelAcquireWithConfig 共用。
581 7 : static HcclResult PrepareV2ChannelAcquire(hccl::hcclComm* hcclComm, HcclComm comm, CommEngine engine)
582 : {
583 7 : hccl::CollComm* collComm = hcclComm->GetCollComm();
584 7 : CHK_PTR_NULL(collComm);
585 7 : hccl::MyRank* myRank = collComm->GetMyRank();
586 7 : CHK_PTR_NULL(myRank);
587 :
588 7 : s32 deviceLogicId = 0;
589 7 : (void)hrtGetDeviceRefresh(&deviceLogicId);
590 7 : u32 rankTableCrc = RankTableCrcBridge::GetInstance().ConsumeRankTableJsonCrc(deviceLogicId);
591 7 : if (rankTableCrc != 0) {
592 0 : CHK_RET(RankConsistencyCheckerV2::GetInstance(deviceLogicId).RecordRankTableCrcV2(rankTableCrc));
593 : }
594 7 : char hcommPkgName[] = "hcomm";
595 7 : char hcommVersionStr[CANN_VERSION_MAX_LEN + 1] = {0};
596 7 : aclError aclRet = aclsysGetVersionStr(hcommPkgName, hcommVersionStr);
597 7 : CHK_PRT_RET(
598 : aclRet != ACL_SUCCESS, HCCL_ERROR("[%s] aclsysGetVersionStr failed, aclRet[%d].", __func__, aclRet),
599 : HCCL_E_INTERNAL);
600 7 : std::string curVersion(hcommVersionStr);
601 7 : CHK_RET(RankConsistencyCheckerV2::GetInstance(deviceLogicId).RecordCannVersionV2(curVersion));
602 :
603 7 : const uint32_t opExpansionMode = myRank->GetOpExpansionMode();
604 7 : if (!CheckCommEngine(engine, opExpansionMode)) {
605 0 : HCCL_ERROR(
606 : "[%s] opExpansionMode[%d] not supported by engine[%s].", __func__, opExpansionMode,
607 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str());
608 0 : return HCCL_E_PARA;
609 : }
610 :
611 7 : if (!GetDebugConfigInited()) {
612 1 : InitDebugConfigByEnv();
613 : }
614 :
615 7 : if (engine != CommEngine::COMM_ENGINE_CPU) {
616 5 : HcclResult monRet = RegisterToClusterMonitor(comm);
617 5 : CHK_PRT_RET(
618 : monRet != HCCL_SUCCESS,
619 : HCCL_ERROR(
620 : "[%s] RegisterToClusterMonitor failed, group[%s], ret[%d].", __func__,
621 : hcclComm->GetIdentifier().c_str(), monRet),
622 : monRet);
623 : }
624 :
625 5 : return HCCL_SUCCESS;
626 7 : }
627 :
628 : // V2 通信域 channel acquire 公共后置处理：symmetric remoteMem 回填、CPU DFX callback、AICPU ReportKernel。
629 : // 非共享路径 HcclChannelAcquire 与共享路径 HcclChannelAcquireWithConfig 共用。
630 4 : static HcclResult FinalizeV2ChannelAcquire(
631 : hccl::hcclComm* hcclComm, CommEngine engine, const std::vector<HcclChannelDesc>& channelDescFinals,
632 : ChannelHandle* channels, uint32_t channelNum, bool hasSymmetricMemHandles, u64 beginTime)
633 : {
634 4 : hccl::CollComm* collComm = hcclComm->GetCollComm();
635 4 : CHK_PTR_NULL(collComm);
636 :
637 4 : if (hasSymmetricMemHandles) {
638 0 : hccl::MyRank* myRank = collComm->GetMyRank();
639 0 : CHK_PTR_NULL(myRank);
640 0 : CHK_RET(UpdateSymmetricRemoteMems(collComm, myRank, channelDescFinals, channels, channelNum));
641 : }
642 :
643 4 : if (engine == COMM_ENGINE_CPU) {
644 2 : HcclCommDfx* hcclCommDfx = collComm->GetHcclCommDfx();
645 2 : CHK_PTR_NULL(hcclCommDfx);
646 2 : auto callback = hcclCommDfx->GetDpuCallback();
647 4 : for (uint32_t idx = 0; idx < channelNum; idx++) {
648 2 : int32_t dpuRet = HcommDpuChannelRegisterDfx(channels[idx], callback);
649 2 : CHK_PRT_RET(
650 : dpuRet != HCCL_SUCCESS,
651 : HCCL_ERROR("[%s] Failed to register DFX callback for channel[%u], ret[%d].", __func__, idx, dpuRet),
652 : static_cast<HcclResult>(dpuRet));
653 : }
654 2 : }
655 :
656 4 : if (IsAicpuEngine(engine)) {
657 1 : HcclCommDfx* hcclCommDfx = collComm->GetHcclCommDfx();
658 1 : CHK_PTR_NULL(hcclCommDfx);
659 1 : std::string kernelName = "RunAicpuIndOpChannelInitV2";
660 : HcclResult reportRet
661 1 : = hcclCommDfx->ReportKernel(beginTime, hcclComm->GetIdentifier(), kernelName, SalGetTid(), false);
662 1 : CHK_PRT_RET(
663 : reportRet != HCCL_SUCCESS,
664 : HCCL_ERROR("[%s] ReportKernel failed, kernelName[%s], ret[%d].", __func__, kernelName.c_str(), reportRet),
665 : reportRet);
666 1 : }
667 :
668 4 : return HCCL_SUCCESS;
669 : }
670 :
671 13 : HcclResult HcclChannelAcquire(
672 : HcclComm comm, CommEngine engine, const HcclChannelDesc* channelDescs, uint32_t channelNum, ChannelHandle* channels)
673 : {
674 13 : HcclUs startut = TIME_NOW();
675 13 : u64 beginTime = Hccl::DlProfFunction::GetInstance().dlMsprofSysCycleTime();
676 : EXCEPTION_HANDLE_BEGIN
677 :
678 : // 入参校验
679 21 : CHK_PTR_NULL(comm);
680 12 : CHK_PTR_NULL(channelDescs);
681 12 : CHK_PTR_NULL(channels);
682 12 : CHK_PRT_RET(
683 : (channelNum == 0 || channelNum > CHANNEL_NUM_MAX),
684 : HCCL_ERROR(
685 : "[%s]Invalid channelNum, channelNum[%u], max channel num[%u]", __func__, channelNum, CHANNEL_NUM_MAX),
686 : HCCL_E_PARA);
687 :
688 12 : HcclResult ret = HCCL_SUCCESS;
689 12 : hccl::hcclComm* hcclComm = static_cast<hccl::hcclComm*>(comm);
690 12 : HCCL_RUN_INFO(
691 : "Entry-%s channelNum[%u], engine[%s] group[%s]", __func__, channelNum,
692 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), hcclComm->GetIdentifier().c_str());
693 12 : std::vector<HcclChannelDesc> channelDescFinals;
694 12 : std::vector<std::vector<HcclMemHandle>> mergedMemHandles;
695 20 : for (uint32_t idx = 0; idx < channelNum; idx++) {
696 : HcclChannelDesc channelDescFinal;
697 12 : HcclChannelDescInit(&channelDescFinal, 1);
698 12 : ret = ProcessHcclResPackReq(channelDescs[idx], channelDescFinal, hcclComm);
699 12 : CHK_PRT_RET(
700 : ret != HCCL_SUCCESS,
701 : HCCL_ERROR(
702 : "ProcessHcclResPackReq failed. channelDesc idx[%u], group[%s], engine[%s] channelNum[%u], ret[%d]", idx,
703 : hcclComm->GetIdentifier().c_str(), GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(),
704 : channelNum, ret),
705 : ret);
706 8 : channelDescFinals.push_back(channelDescFinal);
707 : }
708 :
709 8 : if (hcclComm->IsCommunicatorV2()) { // A5
710 7 : const std::string& commTag = hcclComm->GetIdentifier();
711 7 : hccl::CollComm* collComm = hcclComm->GetCollComm();
712 7 : CHK_PTR_NULL(collComm);
713 :
714 7 : CHK_RET(PrepareV2ChannelAcquire(hcclComm, comm, engine));
715 :
716 5 : bool hasSymmetricMemHandles = false;
717 5 : if (IsAicpuEngine(engine)) {
718 2 : CHK_RET(AppendSymmetricMemHandles(collComm, channelDescFinals, mergedMemHandles, hasSymmetricMemHandles));
719 : }
720 5 : HCCL_INFO(
721 : "[HcclChannelAcquire] AppendSymmetricMemHandles done, group[%s], engine[%d], channelNum[%u], "
722 : "hasSymmetricMemHandles[%d], mergedMemHandleGroups[%zu].",
723 : commTag.c_str(), engine, channelNum, hasSymmetricMemHandles, mergedMemHandles.size());
724 :
725 5 : hccl::MyRank* myRank = collComm->GetMyRank();
726 5 : CHK_PTR_NULL(myRank);
727 5 : ret = myRank->CreateChannels(engine, commTag, channelDescFinals.data(), channelNum, channels);
728 5 : CHK_PRT_RET(
729 : (ret == HCCL_E_AGAIN || ret == HCCL_E_UNAVAIL),
730 : HCCL_WARNING(
731 : "CreateChannels group[%s], engine[%s] ret[%d]", commTag.c_str(),
732 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), ret),
733 : ret);
734 5 : CHK_PRT_RET(
735 : ret != HCCL_SUCCESS,
736 : HCCL_ERROR(
737 : "CreateChannels failed. group[%s], engine[%s] ret[%d]", commTag.c_str(),
738 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), ret),
739 : ret);
740 :
741 4 : CHK_RET(FinalizeV2ChannelAcquire(
742 : hcclComm, engine, channelDescFinals, channels, channelNum, hasSymmetricMemHandles, beginTime));
743 7 : } else {
744 1 : hccl::CollComm* collComm = hcclComm->GetCollComm();
745 1 : if (collComm != nullptr) {
746 0 : hccl::MyRank* myRank = collComm->GetMyRank();
747 0 : if (hcclComm->GetConnectMode() != 0 && engine == COMM_ENGINE_CPU && myRank != nullptr) {
748 0 : const std::string& commTag = hcclComm->GetIdentifier();
749 0 : ret = myRank->CreateChannels(engine, commTag, channelDescFinals.data(), channelNum, channels);
750 0 : } else {
751 0 : auto& channelMgr = hcclComm->GetIndependentOp().GetChannelManager();
752 0 : ret = channelMgr.ChannelCommCreate(
753 0 : hcclComm->GetIdentifier(), engine, channelDescFinals.data(), channelNum, channels);
754 : }
755 : } else {
756 1 : auto& channelMgr = hcclComm->GetIndependentOp().GetChannelManager();
757 1 : ret = channelMgr.ChannelCommCreate(
758 2 : hcclComm->GetIdentifier(), engine, channelDescFinals.data(), channelNum, channels);
759 : }
760 : }
761 :
762 5 : CHK_PRT_RET(
763 : ret != HCCL_SUCCESS,
764 : HCCL_ERROR(
765 : "[%s] Failed to acquire channel, group[%s], engine[%s], channelNum[%u], ret[%d]", __func__,
766 : hcclComm->GetIdentifier().c_str(), GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), channelNum,
767 : ret),
768 : ret);
769 :
770 4 : CHK_RET(ConvertAivChannelHandlesToDevicePtrs(engine, channelDescFinals.data(), channelNum, channels));
771 :
772 4 : HCCL_RUN_INFO(
773 : "[%s] acquire channel success, group[%s], engine[%s], channelNum[%u], take time [%lld]us.", __func__,
774 : hcclComm->GetIdentifier().c_str(), GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), channelNum,
775 : DURATION_US(TIME_NOW() - startut).count());
776 20 : EXCEPTION_HANDLE_END
777 4 : return HCCL_SUCCESS;
778 : }
779 :
780 0 : HcclResult HcclGroupStart() { return HcclLegacyGroupStart(); }
781 :
782 0 : HcclResult HcclGroupEndV2()
783 : {
784 0 : CHK_RET(groupLaunchA5());
785 0 : HCCL_INFO("[GroupEnd] to the end");
786 0 : return HCCL_SUCCESS;
787 : }
788 :
789 0 : HcclResult HcclGroupEnd()
790 : {
791 0 : if (hcclGroupDepth == 0) {
792 0 : HCCL_ERROR("HcclGroupEnd: not in a group call. Didn't call HcclGroupStart before.");
793 0 : return HCCL_E_NOT_SUPPORT;
794 : }
795 0 : if (--hcclGroupDepth > 0) {
796 0 : return HCCL_SUCCESS;
797 : }
798 :
799 0 : HCCL_INFO("[HcclGroupEnd] hcclGroupDepth=[%d]", hcclGroupDepth);
800 : /*遇到最后一个HcclGroupEnd才处理group内的所有任务*/
801 0 : HCCLV2_FUNC_RUN([&]() -> HcclResult {
802 : CHK_RET(HcclLegacyAsyncJobLaunch());
803 : return HcclGroupEndV2();
804 : }());
805 0 : return HcclLegacyGroupEnd();
806 : }
807 :
808 0 : HcclResult HcclGroupStatusGet(bool* isGroupEnabled)
809 : {
810 0 : CHK_PTR_NULL(isGroupEnabled);
811 0 : *isGroupEnabled = (hcclGroupDepth > 0);
812 0 : return HCCL_SUCCESS;
813 : }
814 :
815 0 : static bool IsSharedQueueUbProtocol(CommProtocol protocol)
816 : {
817 0 : return protocol == COMM_PROTOCOL_UB_CTP || protocol == COMM_PROTOCOL_UBC_TP;
818 : }
819 :
820 0 : static bool IsSameLocalEndpoint(const EndpointDesc& a, const EndpointDesc& b)
821 : {
822 0 : return a.protocol == b.protocol && a.commAddr.type == b.commAddr.type
823 0 : && std::memcmp(a.commAddr.raws, b.commAddr.raws, sizeof(a.commAddr.raws)) == 0
824 0 : && a.loc.locType == b.loc.locType && std::memcmp(a.loc.raws, b.loc.raws, sizeof(a.loc.raws)) == 0;
825 : }
826 :
827 0 : static HcclResult ValidateSharedQueueDescs(const std::vector<HcclChannelDesc>& channelDescs)
828 : {
829 0 : for (uint32_t i = 0; i < channelDescs.size(); ++i) {
830 0 : if (!IsSharedQueueUbProtocol(channelDescs[i].channelProtocol)) {
831 0 : HCCL_ERROR(
832 : "[%s] IS_SHARED_QUEUE only supports UB protocols (UB_CTP/UBC_TP), "
833 : "channelDesc[%u] protocol[%d].",
834 : __func__, i, channelDescs[i].channelProtocol);
835 0 : return HCCL_E_NOT_SUPPORT;
836 : }
837 : }
838 :
839 0 : if (channelDescs.size() > 1) {
840 0 : const EndpointDesc& firstLocal = channelDescs[0].localEndpoint;
841 0 : for (uint32_t i = 1; i < channelDescs.size(); ++i) {
842 0 : if (!IsSameLocalEndpoint(firstLocal, channelDescs[i].localEndpoint)) {
843 0 : HCCL_ERROR(
844 : "[%s] all channelDescs must have the same localEndpoint for shared jetty, "
845 : "channelDesc[0] != channelDesc[%u].",
846 : __func__, i);
847 0 : return HCCL_E_PARA;
848 : }
849 : }
850 : }
851 0 : return HCCL_SUCCESS;
852 : }
853 :
854 : struct SharedJettyRemoteGroup {
855 : EndpointDesc remoteEp;
856 : std::vector<uint32_t> descIndices;
857 : };
858 :
859 0 : static HcclResult RegisterMemForSharedJettyChannels(
860 : hccl::MyRank* myRank, EndpointHandle epHandle, std::vector<HcclChannelDesc>& channelDescs,
861 : std::vector<std::vector<MemHandle>>& memHandleStorage)
862 : {
863 0 : uint32_t channelNum = static_cast<uint32_t>(channelDescs.size());
864 0 : for (uint32_t i = 0; i < channelNum; ++i) {
865 0 : CHK_RET(myRank->PrepareMemHandles(
866 : epHandle, channelDescs[i].memHandles, channelDescs[i].memHandleNum, memHandleStorage[i]));
867 0 : channelDescs[i].memHandles = memHandleStorage[i].data();
868 0 : channelDescs[i].memHandleNum = static_cast<uint32_t>(memHandleStorage[i].size());
869 : }
870 0 : return HCCL_SUCCESS;
871 : }
872 :
873 0 : static void GroupChannelDescsByRemoteEp(
874 : const std::vector<HcclChannelDesc>& channelDescs, std::vector<SharedJettyRemoteGroup>& groups)
875 : {
876 0 : auto FindGroup = [&groups](const EndpointDesc& remoteEp) -> SharedJettyRemoteGroup* {
877 0 : for (auto& g : groups) {
878 0 : if (g.remoteEp.protocol == remoteEp.protocol && g.remoteEp.commAddr.type == remoteEp.commAddr.type
879 0 : && std::memcmp(g.remoteEp.commAddr.raws, remoteEp.commAddr.raws, sizeof(remoteEp.commAddr.raws)) == 0
880 0 : && g.remoteEp.loc.locType == remoteEp.loc.locType
881 0 : && std::memcmp(g.remoteEp.loc.raws, remoteEp.loc.raws, sizeof(remoteEp.loc.raws)) == 0) {
882 0 : return &g;
883 : }
884 : }
885 0 : return nullptr;
886 0 : };
887 0 : for (uint32_t i = 0; i < channelDescs.size(); ++i) {
888 0 : const EndpointDesc& remoteEp = channelDescs[i].remoteEndpoint;
889 0 : SharedJettyRemoteGroup* g = FindGroup(remoteEp);
890 0 : if (g == nullptr) {
891 0 : groups.push_back({remoteEp, {i}});
892 : } else {
893 0 : g->descIndices.push_back(i);
894 : }
895 : }
896 0 : }
897 :
898 0 : static HcclResult CreateSharedJettyChannelsForGroup(
899 : CommEngine engine, EndpointHandle epHandle, const std::vector<HcclChannelDesc>& channelDescs, uint32_t repIdx,
900 : const std::string& commTag, hccl::MyRank* myRank, uint32_t needCreate, ChannelHandle* outCh)
901 : {
902 0 : std::vector<HcclChannelDesc> hcclDescs(needCreate, channelDescs[repIdx]);
903 0 : std::vector<HcommChannelDesc> hcommDescs(needCreate);
904 0 : for (uint32_t j = 0; j < needCreate; ++j) {
905 0 : hcommDescs[j] = MyRankUtils::ChannelDescHccl2Hcomm(hcclDescs[j], hccl::CommConfig{});
906 0 : hcommDescs[j].channelName = commTag.c_str();
907 : }
908 0 : std::string socketTag = commTag + "_engine_" + std::to_string(static_cast<uint32_t>(engine));
909 0 : HcclResult sockRet = myRank->BatchCreateSockets(hcclDescs.data(), needCreate, socketTag, hcommDescs);
910 0 : CHK_PRT_RET(
911 : sockRet != HCCL_SUCCESS,
912 : HCCL_ERROR(
913 : "[%s] BatchCreateSockets failed, repIdx[%u], remoteRank[%u], ret[%d].", __func__, repIdx,
914 : channelDescs[repIdx].remoteRank, sockRet),
915 : sockRet);
916 0 : HCCL_INFO("[%s] shared jetty sockets created, repIdx[%u], needCreate[%u].", __func__, repIdx, needCreate);
917 :
918 0 : HcommChannelConfig hcommConfig = nullptr;
919 0 : HcclResult cfgRet = static_cast<HcclResult>(hcomm::ChannelConfigCreate(&hcommConfig));
920 0 : CHK_PRT_RET(
921 : cfgRet != HCCL_SUCCESS, HCCL_ERROR("[%s] ChannelConfigCreate failed, ret[%d].", __func__, cfgRet), cfgRet);
922 0 : auto* hcommCfg = static_cast<hcomm::HcommChannelConfigData*>(hcommConfig);
923 0 : hcommCfg->isSharedQueue = true;
924 :
925 0 : uint32_t created = 0;
926 0 : for (uint32_t j = 0; j < needCreate; ++j) {
927 : HcclResult ret = static_cast<HcclResult>(
928 0 : HcommChannelCreateWithConfig(epHandle, engine, &hcommDescs[j], 1, hcommConfig, &outCh[j]));
929 0 : if (ret != HCCL_SUCCESS) {
930 0 : if (created > 0) {
931 0 : (void)HcommChannelDestroy(outCh, created);
932 : }
933 0 : HCCL_ERROR("[%s] HcommChannelCreateWithConfig failed, j[%u], ret[%d].", __func__, j, ret);
934 0 : (void)hcomm::ChannelConfigDestroy(hcommConfig);
935 0 : return ret;
936 : }
937 0 : created++;
938 : }
939 0 : (void)hcomm::ChannelConfigDestroy(hcommConfig);
940 0 : return HCCL_SUCCESS;
941 0 : }
942 :
943 0 : static HcclResult AcquireSharedJettyGroupChannels(
944 : HcclComm comm, CommEngine engine, const std::vector<HcclChannelDesc>& channelDescs,
945 : const SharedJettyRemoteGroup& group, EndpointHandle epHandle, const std::string& commTag,
946 : const std::string& sharedTag, hccl::MyRank* myRank, const EndpointDesc& localEp, ChannelHandle* channels,
947 : std::vector<bool>* outIsNewChannel)
948 : {
949 : (void)comm;
950 0 : uint32_t requestedNum = static_cast<uint32_t>(group.descIndices.size());
951 0 : hccl::EndpointDescPair epPair = std::make_pair(localEp, group.remoteEp);
952 0 : uint32_t repIdx = group.descIndices[0];
953 :
954 0 : auto createFunc = [engine, &channelDescs, repIdx, epHandle, &commTag,
955 : myRank](uint32_t needCreate, ChannelHandle* outCh) -> HcclResult {
956 0 : return CreateSharedJettyChannelsForGroup(
957 0 : engine, epHandle, channelDescs, repIdx, commTag, myRank, needCreate, outCh);
958 0 : };
959 :
960 0 : std::vector<ChannelHandle> groupOut(requestedNum, 0);
961 0 : uint32_t reusedCount = 0;
962 0 : HcclResult acqRet = hccl::SharedJettyChannelPool::GetInstance().AcquireChannels(
963 : myRank, sharedTag, epPair, requestedNum, createFunc, groupOut.data(), &reusedCount);
964 0 : if (acqRet != HCCL_SUCCESS) {
965 0 : HCCL_ERROR("[%s] AcquireChannels failed for group, ret[%d].", __func__, acqRet);
966 0 : return acqRet;
967 : }
968 :
969 : // 池返回的 handle 按组内 descIndices 回填到 channels 的原位置
970 0 : for (uint32_t k = 0; k < requestedNum; ++k) {
971 0 : uint32_t descIdx = group.descIndices[k];
972 0 : channels[descIdx] = groupOut[k];
973 : // k >= reusedCount 的为新建 channel，回滚时需销毁并从池移除；
974 : // 复用的 channel 仍由池和其他调用方持有，不可销毁
975 0 : if (outIsNewChannel != nullptr && k >= reusedCount) {
976 0 : (*outIsNewChannel)[descIdx] = true;
977 : }
978 0 : u32 remoteRank = channelDescs[descIdx].remoteRank;
979 0 : HcclCommDfx::AddChannelRemoteRankId(commTag, static_cast<u64>(groupOut[k]), remoteRank);
980 : }
981 0 : return HCCL_SUCCESS;
982 0 : }
983 :
984 0 : static void RollbackAcquiredSharedJettyChannels(
985 : uint32_t channelNum, ChannelHandle* channels, const std::vector<bool>* isNewChannel, const EndpointDesc& localEp,
986 : const std::vector<HcclChannelDesc>& channelDescs, hccl::MyRank* myRank, const std::string& sharedTag)
987 : {
988 : // 多组部分失败时回滚已成功的新建 channel
989 : // 复用的 channel 仍由池和其他调用方持有，不可销毁，否则导致 use-after-free
990 0 : for (uint32_t i = 0; i < channelNum; ++i) {
991 0 : if (channels[i] != 0 && isNewChannel != nullptr && (*isNewChannel)[i]) {
992 0 : (void)HcommChannelDestroy(&channels[i], 1);
993 0 : hccl::EndpointDescPair epPair = std::make_pair(localEp, channelDescs[i].remoteEndpoint);
994 0 : hccl::SharedJettyChannelPool::GetInstance().RemoveChannels(myRank, sharedTag, epPair, &channels[i], 1);
995 0 : channels[i] = 0;
996 : }
997 : }
998 0 : }
999 :
1000 0 : static HcclResult AcquireSharedJettyChannels(
1001 : HcclComm comm, CommEngine engine, std::vector<HcclChannelDesc>& channelDescs,
1002 : const hccl::HcclChannelConfigData* cfg, ChannelHandle* channels, std::vector<bool>* outIsNewChannel)
1003 : {
1004 0 : hccl::hcclComm* hcclComm = static_cast<hccl::hcclComm*>(comm);
1005 0 : hccl::CollComm* collComm = hcclComm->GetCollComm();
1006 0 : CHK_PTR_NULL(collComm);
1007 0 : hccl::MyRank* myRank = collComm->GetMyRank();
1008 0 : CHK_PTR_NULL(myRank);
1009 :
1010 0 : const std::string& commTag = hcclComm->GetIdentifier();
1011 0 : const std::string& sharedTag = cfg->sharedQueueTag;
1012 0 : uint32_t channelNum = static_cast<uint32_t>(channelDescs.size());
1013 :
1014 0 : if (outIsNewChannel != nullptr) {
1015 0 : outIsNewChannel->assign(channelNum, false);
1016 : }
1017 :
1018 0 : const EndpointDesc& localEp = channelDescs[0].localEndpoint;
1019 0 : EndpointHandle epHandle = nullptr;
1020 0 : hcomm::EndpointMgr* endpointMgr = myRank->GetEndpointMgr();
1021 0 : CHK_PTR_NULL(endpointMgr);
1022 0 : CHK_RET(endpointMgr->Get(localEp, epHandle));
1023 :
1024 : // memHandleStorage 持有 memHandleVec 的生命周期，确保 channelDescs[].memHandles 在本函数内有效。
1025 : // 无论 memVec 是否为空都执行 RegisterMemory 并覆盖 memHandles：
1026 : // 空时 memHandleStorage[i] 为空 → memHandles=nullptr/memHandleNum=0，避免残留用户传入的无效句柄。
1027 0 : std::vector<std::vector<MemHandle>> memHandleStorage(channelNum);
1028 0 : CHK_RET(RegisterMemForSharedJettyChannels(myRank, epHandle, channelDescs, memHandleStorage));
1029 :
1030 0 : std::vector<SharedJettyRemoteGroup> groups;
1031 0 : GroupChannelDescsByRemoteEp(channelDescs, groups);
1032 :
1033 0 : HcclResult groupRet = HCCL_SUCCESS;
1034 0 : for (const auto& group : groups) {
1035 0 : groupRet = AcquireSharedJettyGroupChannels(
1036 : comm, engine, channelDescs, group, epHandle, commTag, sharedTag, myRank, localEp, channels,
1037 : outIsNewChannel);
1038 0 : if (groupRet != HCCL_SUCCESS) {
1039 0 : break;
1040 : }
1041 : }
1042 :
1043 0 : if (groupRet != HCCL_SUCCESS) {
1044 0 : RollbackAcquiredSharedJettyChannels(
1045 : channelNum, channels, outIsNewChannel, localEp, channelDescs, myRank, sharedTag);
1046 0 : return groupRet;
1047 : }
1048 :
1049 0 : HCCL_INFO(
1050 : "[%s] shared jetty channels acquired, comm[%p], tag[%s], channelNum[%u], remoteGroups[%zu].", __func__, comm,
1051 : sharedTag.c_str(), channelNum, groups.size());
1052 :
1053 : // memHandleStorage 即将析构，清空 channelDescs 中的悬空指针，防止调用方误用
1054 0 : for (uint32_t i = 0; i < channelNum; ++i) {
1055 0 : channelDescs[i].memHandles = nullptr;
1056 0 : channelDescs[i].memHandleNum = 0;
1057 : }
1058 0 : return HCCL_SUCCESS;
1059 0 : }
1060 :
1061 0 : static HcclResult ParseSharedQueueConfig(
1062 : HcclChannelConfig config, CommEngine engine, HcclComm comm, bool& isSharedQueue, std::string& sharedQueueTag,
1063 : hccl::hcclComm*& hcclComm)
1064 : {
1065 0 : isSharedQueue = false;
1066 0 : if (config != nullptr) {
1067 0 : auto* cfg = static_cast<hccl::HcclChannelConfigData*>(config);
1068 0 : isSharedQueue = cfg->isSharedQueue;
1069 0 : sharedQueueTag = cfg->sharedQueueTag;
1070 : }
1071 :
1072 0 : if (!isSharedQueue) {
1073 0 : return HCCL_SUCCESS;
1074 : }
1075 :
1076 0 : if (sharedQueueTag.empty()) {
1077 0 : HCCL_ERROR("[%s] SHARED_QUEUE_TAG must be set when IS_SHARED_QUEUE is true.", __func__);
1078 0 : return HCCL_E_PARA;
1079 : }
1080 :
1081 0 : if (engine != COMM_ENGINE_AIV) {
1082 0 : HCCL_ERROR(
1083 : "[%s] IS_SHARED_QUEUE currently only supports AIV engine, engine[%d].", __func__, static_cast<int>(engine));
1084 0 : return HCCL_E_NOT_SUPPORT;
1085 : }
1086 :
1087 0 : hcclComm = static_cast<hccl::hcclComm*>(comm);
1088 0 : if (!hcclComm->IsCommunicatorV2()) {
1089 0 : HCCL_ERROR("[%s] IS_SHARED_QUEUE only supports V2 communicator.", __func__);
1090 0 : return HCCL_E_NOT_SUPPORT;
1091 : }
1092 0 : return HCCL_SUCCESS;
1093 : }
1094 :
1095 0 : static HcclResult ProcessSharedQueueDescFinals(
1096 : const HcclChannelDesc* channelDescs, uint32_t channelNum, hccl::hcclComm* hcclComm, CommEngine engine,
1097 : std::vector<HcclChannelDesc>& channelDescFinals)
1098 : {
1099 0 : for (uint32_t idx = 0; idx < channelNum; idx++) {
1100 : HcclChannelDesc channelDescFinal;
1101 0 : HcclChannelDescInit(&channelDescFinal, 1);
1102 0 : HcclResult ret = ProcessHcclResPackReq(channelDescs[idx], channelDescFinal, hcclComm);
1103 0 : CHK_PRT_RET(
1104 : ret != HCCL_SUCCESS,
1105 : HCCL_ERROR(
1106 : "ProcessHcclResPackReq failed. channelDesc idx[%u], group[%s], engine[%s] "
1107 : "channelNum[%u], ret[%d]",
1108 : idx, hcclComm->GetIdentifier().c_str(), GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(),
1109 : channelNum, ret),
1110 : ret);
1111 0 : channelDescFinals.push_back(channelDescFinal);
1112 : }
1113 0 : return HCCL_SUCCESS;
1114 : }
1115 :
1116 0 : static void DestroyAndClearSharedJettyChannels(
1117 : hccl::hcclComm* hcclComm, const std::string& sharedQueueTag, uint32_t channelNum, ChannelHandle* channels,
1118 : const std::vector<bool>& isNewChannel, const std::vector<ChannelHandle>& channelsCopy,
1119 : const std::vector<HcclChannelDesc>& channelDescFinals)
1120 : {
1121 : // 仅销毁本轮新建的 channel，复用的 channel 保留在池中供其他调用方使用
1122 0 : for (uint32_t i = 0; i < channelNum; ++i) {
1123 0 : if (channels[i] != 0 && isNewChannel[i]) {
1124 0 : (void)HcommChannelDestroy(&channels[i], 1);
1125 0 : channels[i] = 0;
1126 : }
1127 : }
1128 : // 从池中移除已销毁的新建句柄，避免重试时返回已销毁的 channel
1129 0 : hccl::CollComm* collComm = hcclComm->GetCollComm();
1130 0 : if (collComm == nullptr) {
1131 0 : return;
1132 : }
1133 0 : hccl::MyRank* myRank = collComm->GetMyRank();
1134 0 : if (myRank == nullptr) {
1135 0 : return;
1136 : }
1137 0 : const EndpointDesc& localEp = channelDescFinals[0].localEndpoint;
1138 0 : for (uint32_t i = 0; i < channelNum; ++i) {
1139 0 : if (channelsCopy[i] == 0 || !isNewChannel[i]) {
1140 0 : continue;
1141 : }
1142 0 : const EndpointDesc& remoteEp = channelDescFinals[i].remoteEndpoint;
1143 0 : hccl::EndpointDescPair epPair = std::make_pair(localEp, remoteEp);
1144 0 : hccl::SharedJettyChannelPool::GetInstance().RemoveChannels(myRank, sharedQueueTag, epPair, &channelsCopy[i], 1);
1145 : }
1146 : }
1147 :
1148 : static HcclResult
1149 0 : WaitForSharedJettyChannelsReady(uint32_t channelNum, ChannelHandle* channels, hccl::hcclComm* hcclComm)
1150 : {
1151 0 : std::vector<int32_t> statusList(channelNum, 0);
1152 0 : auto linkTimeout = std::chrono::seconds(Hccl::EnvConfig::GetInstance().GetSocketConfig().GetLinkTimeOut());
1153 0 : auto startTime = std::chrono::steady_clock::now();
1154 : while (true) {
1155 0 : HcclResult statusRet = static_cast<HcclResult>(HcommChannelGetStatus(channels, channelNum, statusList.data()));
1156 0 : if (statusRet != HCCL_SUCCESS && statusRet != HCCL_E_AGAIN) {
1157 0 : HCCL_ERROR("[%s] HcommChannelGetStatus failed during shared jetty connect, ret[%d].", __func__, statusRet);
1158 0 : return statusRet;
1159 : }
1160 0 : bool allReady = true;
1161 0 : for (uint32_t i = 0; i < channelNum; ++i) {
1162 0 : if (statusList[i] == hcomm::HCOMM_CHANNEL_STATUS_FAILED
1163 0 : || statusList[i] == hcomm::HCOMM_CHANNEL_STATUS_TIMEOUT) {
1164 0 : HCCL_ERROR("[%s] shared jetty channel[%u] connect failed, status[%d].", __func__, i, statusList[i]);
1165 0 : return HCCL_E_NETWORK;
1166 : }
1167 0 : if (statusList[i] != hcomm::HCOMM_CHANNEL_STATUS_READY) {
1168 0 : allReady = false;
1169 : }
1170 : }
1171 0 : if (allReady) {
1172 0 : return HCCL_SUCCESS;
1173 : }
1174 0 : if ((std::chrono::steady_clock::now() - startTime) >= linkTimeout) {
1175 0 : HCCL_ERROR(
1176 : "[%s] shared jetty channel connect timeout, group[%s].", __func__, hcclComm->GetIdentifier().c_str());
1177 0 : return HCCL_E_TIMEOUT;
1178 : }
1179 0 : std::this_thread::sleep_for(std::chrono::milliseconds(2));
1180 0 : }
1181 0 : }
1182 :
1183 0 : static HcclResult ExchangeConsistencyForSharedJetty(
1184 : hccl::hcclComm* hcclComm, CommEngine engine, uint32_t channelNum,
1185 : const std::vector<HcclChannelDesc>& channelDescFinals, const std::vector<bool>& isNewChannel)
1186 : {
1187 0 : hccl::CollComm* collComm = hcclComm->GetCollComm();
1188 0 : CHK_PTR_NULL(collComm);
1189 0 : hccl::MyRank* myRank = collComm->GetMyRank();
1190 0 : CHK_PTR_NULL(myRank);
1191 :
1192 0 : std::vector<HcommChannelDesc> consistencyDescs(channelNum);
1193 0 : for (uint32_t i = 0; i < channelNum; ++i) {
1194 0 : consistencyDescs[i] = MyRankUtils::ChannelDescHccl2Hcomm(channelDescFinals[i], hccl::CommConfig{});
1195 0 : consistencyDescs[i].channelName = hcclComm->GetIdentifier().c_str();
1196 : }
1197 :
1198 : std::string consistencySocketTag
1199 0 : = hcclComm->GetIdentifier() + "_engine_" + std::to_string(static_cast<uint32_t>(engine));
1200 : HcclResult sockRet
1201 0 : = myRank->BatchCreateSockets(channelDescFinals.data(), channelNum, consistencySocketTag, consistencyDescs);
1202 0 : CHK_PRT_RET(
1203 : sockRet != HCCL_SUCCESS,
1204 : HCCL_ERROR("[%s] BatchCreateSockets for consistency failed, ret[%d].", __func__, sockRet), sockRet);
1205 :
1206 0 : std::vector<std::pair<u32, u32>> newChannelIdxs;
1207 0 : for (uint32_t i = 0; i < channelNum; ++i) {
1208 0 : if (isNewChannel[i]) {
1209 0 : newChannelIdxs.emplace_back(i, 0U);
1210 : }
1211 : }
1212 0 : HcclResult exchRet = myRank->BatchExchangeAndCheckConsistency(
1213 : channelDescFinals.data(), consistencyDescs, channelNum, newChannelIdxs, engine);
1214 0 : CHK_PRT_RET(
1215 : exchRet != HCCL_SUCCESS,
1216 : HCCL_ERROR(
1217 : "[%s] BatchExchangeAndCheckConsistency failed, group[%s], ret[%d].", __func__,
1218 : hcclComm->GetIdentifier().c_str(), exchRet),
1219 : exchRet);
1220 0 : return HCCL_SUCCESS;
1221 0 : }
1222 :
1223 : // 推进建链状态机至 READY + 一致性交换，失败时销毁已获取的新建 channel 并从池中移除
1224 0 : static HcclResult FinalizeSharedJettyAcquisition(
1225 : hccl::hcclComm* hcclComm, CommEngine engine, uint32_t channelNum, ChannelHandle* channels,
1226 : const std::vector<bool>& isNewChannel, const std::vector<HcclChannelDesc>& channelDescFinals,
1227 : const std::string& sharedQueueTag)
1228 : {
1229 0 : std::vector<ChannelHandle> channelsCopy(channels, channels + channelNum);
1230 :
1231 0 : HcclResult waitRet = WaitForSharedJettyChannelsReady(channelNum, channels, hcclComm);
1232 0 : if (waitRet != HCCL_SUCCESS) {
1233 0 : DestroyAndClearSharedJettyChannels(
1234 : hcclComm, sharedQueueTag, channelNum, channels, isNewChannel, channelsCopy, channelDescFinals);
1235 0 : return waitRet;
1236 : }
1237 :
1238 : HcclResult exchRet
1239 0 : = ExchangeConsistencyForSharedJetty(hcclComm, engine, channelNum, channelDescFinals, isNewChannel);
1240 0 : if (exchRet != HCCL_SUCCESS) {
1241 0 : DestroyAndClearSharedJettyChannels(
1242 : hcclComm, sharedQueueTag, channelNum, channels, isNewChannel, channelsCopy, channelDescFinals);
1243 0 : return exchRet;
1244 : }
1245 0 : return HCCL_SUCCESS;
1246 0 : }
1247 :
1248 0 : HcclResult HcclChannelAcquireWithConfig(
1249 : HcclComm comm, CommEngine engine, const HcclChannelDesc* channelDescs, uint32_t channelNum,
1250 : HcclChannelConfig config, ChannelHandle* channels)
1251 : {
1252 0 : HcclUs startut = TIME_NOW();
1253 : EXCEPTION_HANDLE_BEGIN
1254 :
1255 : // 入参校验
1256 0 : CHK_PTR_NULL(comm);
1257 0 : CHK_PTR_NULL(channelDescs);
1258 0 : CHK_PTR_NULL(channels);
1259 0 : CHK_PRT_RET(
1260 : (channelNum == 0 || channelNum > CHANNEL_NUM_MAX),
1261 : HCCL_ERROR(
1262 : "[%s]Invalid channelNum, channelNum[%u], max channel num[%u]", __func__, channelNum, CHANNEL_NUM_MAX),
1263 : HCCL_E_PARA);
1264 :
1265 0 : bool isSharedQueue = false;
1266 0 : std::string sharedQueueTag;
1267 0 : hccl::hcclComm* hcclComm = nullptr;
1268 0 : CHK_RET(ParseSharedQueueConfig(config, engine, comm, isSharedQueue, sharedQueueTag, hcclComm));
1269 0 : if (!isSharedQueue) {
1270 0 : return HcclChannelAcquire(comm, engine, channelDescs, channelNum, channels);
1271 : }
1272 :
1273 0 : u64 beginTime = Hccl::DlProfFunction::GetInstance().dlMsprofSysCycleTime();
1274 0 : CHK_RET(PrepareV2ChannelAcquire(hcclComm, comm, engine));
1275 :
1276 : // 复用 HcclChannelAcquire 的前置校验（ProcessHcclResPackReq），保证共享/非共享路径校验一致
1277 0 : std::vector<HcclChannelDesc> channelDescFinals;
1278 0 : CHK_RET(ProcessSharedQueueDescFinals(channelDescs, channelNum, hcclComm, engine, channelDescFinals));
1279 0 : CHK_RET(ValidateSharedQueueDescs(channelDescFinals));
1280 :
1281 0 : std::vector<std::vector<HcclMemHandle>> mergedMemHandles;
1282 0 : bool hasSymmetricMemHandles = false;
1283 0 : if (IsAicpuEngine(engine)) {
1284 0 : hccl::CollComm* collComm = hcclComm->GetCollComm();
1285 0 : CHK_PTR_NULL(collComm);
1286 0 : CHK_RET(AppendSymmetricMemHandles(collComm, channelDescFinals, mergedMemHandles, hasSymmetricMemHandles));
1287 : }
1288 :
1289 0 : auto* cfg = static_cast<hccl::HcclChannelConfigData*>(config);
1290 0 : std::vector<bool> isNewChannel;
1291 0 : HcclResult ret = AcquireSharedJettyChannels(comm, engine, channelDescFinals, cfg, channels, &isNewChannel);
1292 0 : CHK_PRT_RET(
1293 : ret != HCCL_SUCCESS, HCCL_ERROR(
1294 : "[%s] AcquireSharedJettyChannels failed, group[%s], ret[%d].", __func__,
1295 : hcclComm->GetIdentifier().c_str(), ret);
1296 : for (uint32_t i = 0; i < channelNum; ++i) { channels[i] = 0; }, ret);
1297 :
1298 : // 推进建链状态机至 READY + 一致性交换，失败时自动清理
1299 0 : CHK_RET(FinalizeSharedJettyAcquisition(
1300 : hcclComm, engine, channelNum, channels, isNewChannel, channelDescFinals, sharedQueueTag));
1301 :
1302 0 : CHK_RET(FinalizeV2ChannelAcquire(
1303 : hcclComm, engine, channelDescFinals, channels, channelNum, hasSymmetricMemHandles, beginTime));
1304 :
1305 0 : HCCL_RUN_INFO(
1306 : "[%s] acquire shared jetty channels success, group[%s], engine[%s], channelNum[%u], take time [%lld]us.",
1307 : __func__, hcclComm->GetIdentifier().c_str(), GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(),
1308 : channelNum, DURATION_US(TIME_NOW() - startut));
1309 0 : EXCEPTION_HANDLE_END
1310 0 : return HCCL_SUCCESS;
1311 : }
|
