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