Line data Source code
1 : /**
2 : * Copyright (c) 2026 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 <cstring>
12 : #include <chrono>
13 : #include <vector>
14 :
15 : #include "hcomm_c_adpt.h"
16 : #include "hcomm_c_adpt_common.h"
17 : #include "hcomm_res.h"
18 : #include "hcomm_result_defs.h"
19 : #include "hcomm_res_defs.h"
20 : #include "hcomm_channel.h"
21 : #include "log.h"
22 : #include "param_check_pub.h"
23 : #include "comm_engine_utils.h"
24 : #include "channel_process.h"
25 : #include "aicpu_ts_channel_helper.h"
26 : #include "channel_config.h"
27 : #include "shared_jetty_mgr.h"
28 : #include "endpoint.h"
29 : #include "builtin_endpoint_ops.h"
30 : #include "nic_plugin_holder.h"
31 : #include "nic_plugin_manager.h"
32 : #include "acl/acl_rt.h"
33 : #include "adapter_rts_common.h"
34 : #include "tp_qos.h"
35 : #include "hccl/hccl_types.h"
36 :
37 : using namespace hcomm;
38 :
39 : constexpr uint32_t kDscpToRoceTcShift = 2U; // RoCE TC = DSCP << 2（DiffServ 高 6 位为 DSCP）
40 :
41 7 : static void ApplyRoceQosCompatToSlTc(HcommChannelDesc& channelDesc)
42 : {
43 7 : if (channelDesc.qos == HCCL_COMM_QOS_CONFIG_NOT_SET) {
44 3 : return;
45 : }
46 :
47 4 : const uint8_t sl = static_cast<uint8_t>(channelDesc.qos & 0xFFU);
48 4 : uint8_t dscp = Hccl::kUboeDefaultDscp;
49 4 : s32 userDevId = 0;
50 4 : s32 phyDevId = 0;
51 4 : if (hrtGetDevice(&userDevId) != HCCL_SUCCESS || aclrtGetPhyDevIdByUserDevId(userDevId, &phyDevId) != ACL_SUCCESS) {
52 0 : HCCL_WARNING(
53 : "[ApplyRoceQosCompatToSlTc] get phyDevId failed, userDevId[%d], fallback to default dscp[%u].", userDevId,
54 : static_cast<unsigned>(dscp));
55 : } else {
56 4 : (void)Hccl::TpQosGetDscpByQosFromHccnCfg(static_cast<uint32_t>(phyDevId), sl, dscp);
57 : }
58 :
59 4 : channelDesc.roceAttr.sl = sl;
60 4 : channelDesc.roceAttr.tc = static_cast<uint8_t>((static_cast<uint32_t>(dscp) << kDscpToRoceTcShift) & 0xFFU);
61 4 : HCCL_INFO(
62 : "[ApplyRoceQosCompatToSlTc] qos compat: qos[%u] userDevId[%d] phyDevId[%d] dscp[%u] sl[%u] tc[%u].",
63 : channelDesc.qos, userDevId, phyDevId, static_cast<unsigned>(dscp),
64 : static_cast<unsigned>(channelDesc.roceAttr.sl), static_cast<unsigned>(channelDesc.roceAttr.tc));
65 : }
66 :
67 : namespace {
68 0 : void DestroyPluginCtx(HcommNicChannelOps* ops, void* pluginCtx)
69 : {
70 0 : if (ops != nullptr && ops->destroy != nullptr) {
71 0 : int32_t ret = ops->destroy(pluginCtx);
72 0 : if (ret != HCCL_SUCCESS) {
73 0 : HCCL_WARNING("[%s] plugin channel destroy failed, ret[%d].", __func__, ret);
74 : }
75 : }
76 0 : }
77 :
78 0 : void RollbackPluginChannels(ChannelHandle* channels, uint32_t count)
79 : {
80 0 : for (uint32_t i = 0; i < count; ++i) {
81 0 : if (channels[i] == 0)
82 0 : continue;
83 0 : auto* ch = CHANNEL_FROM_HANDLE(channels[i]);
84 0 : if (ch != nullptr) {
85 0 : HcclResult ret = ChannelProcess::RemovePluginChannelFromMap(reinterpret_cast<ChannelHandle>(ch));
86 0 : if (ret != HCCL_SUCCESS) {
87 0 : HCCL_WARNING(
88 : "[%s] plugin channel not found in map during rollback, handle[0x%llx], ret[%d].", __func__,
89 : channels[i], ret);
90 : }
91 : }
92 0 : channels[i] = 0;
93 : }
94 0 : }
95 :
96 2 : HcommResult CreateOnePluginChannel(
97 : const NicPluginEntry* entry, void* epCtx, HcommChannelDesc* channelDesc, ChannelHandle* outChannel)
98 : {
99 2 : *outChannel = 0;
100 :
101 2 : void* pluginCtx = nullptr;
102 2 : HcommNicChannelOps* pluginOps = nullptr;
103 2 : HcommResult ret = static_cast<HcommResult>(entry->createChannel(epCtx, channelDesc, &pluginCtx, &pluginOps));
104 2 : CHK_PRT_RET(
105 : (ret != HCCL_SUCCESS), HCCL_ERROR("[NicPlugin][%s] createChannel failed, ret[%d].", __func__, ret), ret);
106 :
107 2 : if (!ValidateChannelOps(pluginOps)) {
108 0 : HCCL_ERROR("[NicPlugin][%s] invalid channel ops.", __func__);
109 0 : DestroyPluginCtx(pluginOps, pluginCtx);
110 0 : return HCCL_E_INTERNAL;
111 : }
112 :
113 2 : HcommNicChannelOps* filledOps = nullptr;
114 2 : ret = FillDefaultChannelOps(pluginOps, &filledOps);
115 2 : if (ret != HCCL_SUCCESS) {
116 0 : HCCL_ERROR("[NicPlugin][%s] FillDefaultChannelOps failed, ret[%d].", __func__, ret);
117 0 : DestroyPluginCtx(pluginOps, pluginCtx);
118 0 : return ret;
119 : }
120 :
121 2 : ret = static_cast<HcommResult>(filledOps->init(pluginCtx));
122 2 : if (ret != HCCL_SUCCESS) {
123 0 : int32_t destroyRet = filledOps->destroy(pluginCtx);
124 0 : if (destroyRet != HCCL_SUCCESS) {
125 0 : HCCL_WARNING("[%s] plugin channel destroy failed after init failure, ret[%d].", __func__, destroyRet);
126 : }
127 0 : delete filledOps;
128 0 : HCCL_ERROR("[NicPlugin][%s] plugin channel init failed, ret[%d].", __func__, ret);
129 0 : return ret;
130 : }
131 :
132 2 : auto holder = std::make_shared<hcomm::PluginChannelHolder>(entry);
133 2 : holder->SetNicChannelCtx(filledOps, pluginCtx);
134 2 : ChannelHandle handle = reinterpret_cast<ChannelHandle>(holder.get());
135 :
136 2 : ret = static_cast<HcommResult>(ChannelProcess::InsertPluginChannelToMap(handle, std::move(holder)));
137 2 : CHK_PRT_RET(
138 : (ret != HCCL_SUCCESS), HCCL_ERROR("[NicPlugin][%s] InsertChannelToMap failed, ret[%d].", __func__, ret), ret);
139 :
140 2 : *outChannel = MAKE_PLUGIN_CH_HANDLE(handle);
141 2 : HCCL_INFO("[%s] plugin channel created, handle[0x%llx].", __func__, handle);
142 2 : return HCCL_SUCCESS;
143 2 : }
144 :
145 : } // namespace
146 :
147 48 : HcommResult CheckUbAttr(HcommChannelDesc& channelDesc, [[maybe_unused]] CommEngine engine)
148 : {
149 48 : if (channelDesc.remoteEndpoint.protocol != COMM_PROTOCOL_UBC_TP
150 47 : && channelDesc.remoteEndpoint.protocol != COMM_PROTOCOL_UBOE
151 47 : && channelDesc.remoteEndpoint.protocol != COMM_PROTOCOL_UB_RTP
152 47 : && channelDesc.remoteEndpoint.protocol != COMM_PROTOCOL_UB_CTP) {
153 24 : return HCCL_SUCCESS;
154 : }
155 :
156 : // 暂不支持UBOE场景下配置SqDepth
157 24 : if (channelDesc.remoteEndpoint.protocol == COMM_PROTOCOL_UBOE) {
158 0 : return HCCL_SUCCESS;
159 : }
160 :
161 : // check sqDepth
162 24 : if (channelDesc.ubAttr.sqDepth == UB_SQ_DEPTH_NOT_SET) {
163 16 : HCCL_INFO("[%s] use default ubAttr.sqDepth.", __func__);
164 16 : return HCCL_SUCCESS;
165 : }
166 :
167 : // channelDesc.ubAttr.sqDepth调整到2的整数次幂
168 8 : auto GetNextPowerOfTwo = [](uint32_t n) -> uint32_t {
169 8 : n--;
170 8 : n |= n >> 1;
171 8 : n |= n >> 2;
172 8 : n |= n >> 4;
173 8 : n |= n >> 8;
174 8 : n |= n >> 16;
175 8 : return n + 1;
176 : };
177 :
178 8 : channelDesc.ubAttr.sqDepth = GetNextPowerOfTwo(channelDesc.ubAttr.sqDepth);
179 :
180 8 : return HCCL_SUCCESS;
181 : }
182 :
183 47 : HcommResult CheckUbMemAttr(HcommChannelDesc& channelDesc)
184 : {
185 47 : if (channelDesc.remoteEndpoint.protocol != COMM_PROTOCOL_UB_MEM) {
186 28 : return HCOMM_SUCCESS;
187 : }
188 :
189 19 : if (channelDesc.ubMemAttr.pathMode == 0xFF) {
190 1 : HCCL_INFO("[%s] use default ubMemAttr.pathMode, set to 0.", __func__);
191 1 : channelDesc.ubMemAttr.pathMode = 0;
192 1 : return HCOMM_SUCCESS;
193 : }
194 :
195 18 : if (channelDesc.ubMemAttr.pathMode > 2) {
196 3 : HCCL_ERROR("[%s] invalid ubMemAttr.pathMode[%u], should be 0 ~ 2.", __func__, channelDesc.ubMemAttr.pathMode);
197 3 : return HCCL_E_PARA;
198 : }
199 15 : return HCOMM_SUCCESS;
200 : }
201 :
202 46 : HcommResult CheckRoceAttr(HcommChannelDesc& channelDesc)
203 : {
204 46 : if (channelDesc.remoteEndpoint.protocol != COMM_PROTOCOL_ROCE) {
205 39 : return HCCL_SUCCESS;
206 : }
207 :
208 7 : if (channelDesc.roceAttr.queueNum == INVALID_UINT) {
209 3 : channelDesc.roceAttr.queueNum = 1;
210 3 : HCCL_INFO("[%s] set roceAttr.queueNum to 1.", __func__);
211 : }
212 :
213 7 : if (channelDesc.roceAttr.cqAttrFlags == INVALID_UINT) {
214 7 : channelDesc.roceAttr.cqAttrFlags = 0;
215 7 : HCCL_INFO("[%s] set roceAttr.cqAttrFlags to 0.", __func__);
216 : }
217 :
218 7 : ApplyRoceQosCompatToSlTc(channelDesc);
219 :
220 7 : return HCCL_SUCCESS;
221 : }
222 :
223 : namespace {
224 40 : void ApplyHcommChannelDescV1Fields(const HcommChannelDesc& channelDesc, HcommChannelDesc& channelDescFinal)
225 : {
226 40 : if (channelDesc.header.version < HCOMM_CHANNEL_VERSION_ONE) {
227 0 : return;
228 : }
229 :
230 40 : channelDescFinal.remoteEndpoint = channelDesc.remoteEndpoint;
231 40 : channelDescFinal.notifyNum = channelDesc.notifyNum;
232 40 : channelDescFinal.exchangeAllMems = channelDesc.exchangeAllMems;
233 40 : channelDescFinal.memHandles = channelDesc.memHandles;
234 40 : channelDescFinal.memHandleNum = channelDesc.memHandleNum;
235 40 : channelDescFinal.socket = channelDesc.socket;
236 40 : channelDescFinal.role = channelDesc.role;
237 40 : channelDescFinal.port = channelDesc.port;
238 : }
239 :
240 40 : HcommResult ProcessHcommChannelDescs(const HcommChannelDesc& channelDesc, HcommChannelDesc& channelDescFinal)
241 : {
242 40 : if (channelDesc.header.size < sizeof(CommAbiHeader)) {
243 0 : HCCL_ERROR("[%s] invalid channelDesc.header.size[%u].", __func__, channelDesc.header.size);
244 0 : return HCCL_E_PARA;
245 : }
246 :
247 40 : if (channelDesc.header.magicWord != channelDescFinal.header.magicWord) {
248 0 : HCCL_ERROR(
249 : "[%s] channelDesc.header.magicWord[0x%08x] is invalid, expected[0x%08x].", __func__,
250 : channelDesc.header.magicWord, channelDescFinal.header.magicWord);
251 0 : return HCCL_E_PARA;
252 : }
253 :
254 40 : const uint32_t copySize = (channelDescFinal.header.size < channelDesc.header.size ? channelDescFinal.header.size :
255 40 : channelDesc.header.size)
256 0 : - sizeof(CommAbiHeader);
257 40 : CHK_SAFETY_FUNC_RET(memcpy_s(
258 : reinterpret_cast<uint8_t*>(&channelDescFinal) + sizeof(CommAbiHeader), copySize,
259 : reinterpret_cast<const uint8_t*>(&channelDesc) + sizeof(CommAbiHeader), copySize));
260 40 : ApplyHcommChannelDescV1Fields(channelDesc, channelDescFinal);
261 40 : if (channelDesc.header.version > HCOMM_CHANNEL_VERSION) {
262 0 : HCCL_RUN_WARNING(
263 : "The version of provided [%u] is higher than the current version[%u], "
264 : "unsupported configuration will be ignored.",
265 : channelDesc.header.version, HCOMM_CHANNEL_VERSION);
266 40 : } else if (channelDesc.header.version < HCOMM_CHANNEL_VERSION) {
267 1 : HCCL_RUN_WARNING(
268 : "The version of provided [%u] is lower than the current version[%u], "
269 : "configurations supported by later versions will be ignored.",
270 : channelDesc.header.version, HCOMM_CHANNEL_VERSION);
271 : }
272 :
273 : // qos：低版本时置默认值
274 40 : if (channelDesc.header.version <= HCOMM_CHANNEL_VERSION_ONE) {
275 1 : channelDescFinal.qos = 0xFFFFFFFFU;
276 : } else {
277 39 : channelDescFinal.qos = channelDesc.qos;
278 : }
279 :
280 : // v3：channelName，低版本时置 NULL
281 40 : constexpr uint32_t HCOMM_CHANNEL_VERSION_THREE = 3U;
282 40 : if (channelDesc.header.version < HCOMM_CHANNEL_VERSION_THREE) {
283 1 : channelDescFinal.channelName = nullptr;
284 : } else {
285 39 : channelDescFinal.channelName = channelDesc.channelName;
286 39 : if (channelDescFinal.channelName != nullptr
287 1 : && reinterpret_cast<uintptr_t>(channelDescFinal.channelName) == static_cast<uintptr_t>(-1)) {
288 0 : channelDescFinal.channelName = nullptr;
289 : }
290 : }
291 :
292 40 : if (channelDescFinal.channelName != nullptr) {
293 1 : size_t nameLen = strnlen(channelDescFinal.channelName, HCOMM_CHANNEL_NAME_MAX_LEN + 1);
294 1 : if (nameLen > HCOMM_CHANNEL_NAME_MAX_LEN) {
295 0 : HCCL_ERROR("[%s] channelName too long, max len[%u].", __func__, HCOMM_CHANNEL_NAME_MAX_LEN);
296 0 : return HCCL_E_PARA;
297 : }
298 : }
299 :
300 : // v4：roceAttr.srcPortList，低版本时 union 内该位置为脏数据，置 NULL
301 40 : if (channelDesc.header.version < HCOMM_CHANNEL_VERSION) {
302 1 : channelDescFinal.roceAttr.srcPortList = nullptr;
303 : } else {
304 39 : channelDescFinal.roceAttr.srcPortList = channelDesc.roceAttr.srcPortList;
305 : }
306 :
307 40 : return HCOMM_SUCCESS;
308 : }
309 :
310 40 : HcommResult NormalizeHcommChannelDescs(
311 : HcommChannelDesc* channelDescs, uint32_t channelNum, std::vector<HcommChannelDesc>& channelDescFinals,
312 : CommEngine engine)
313 : {
314 40 : channelDescFinals.clear();
315 40 : channelDescFinals.reserve(channelNum);
316 80 : for (uint32_t idx = 0; idx < channelNum; ++idx) {
317 40 : HcommChannelDesc channelDescFinal{};
318 40 : HcommResult ret = HcommChannelDescInit(&channelDescFinal, 1);
319 40 : if (ret != HCOMM_SUCCESS) {
320 0 : return ret;
321 : }
322 40 : ret = ProcessHcommChannelDescs(channelDescs[idx], channelDescFinal);
323 40 : if (ret != HCOMM_SUCCESS) {
324 0 : HCCL_ERROR("[%s] failed to normalize channelDesc[%u], ret[%d].", __func__, idx, ret);
325 0 : return ret;
326 : }
327 40 : ret = CheckUbAttr(channelDescFinal, engine);
328 40 : if (ret != HCOMM_SUCCESS) {
329 0 : HCCL_ERROR("[%s] CheckUbAttr failed, ret[%d].", __func__, ret);
330 0 : return ret;
331 : }
332 40 : ret = CheckUbMemAttr(channelDescFinal);
333 40 : if (ret != HCOMM_SUCCESS) {
334 0 : HCCL_ERROR("[%s] CheckUbMemAttr failed, ret[%d].", __func__, ret);
335 0 : return ret;
336 : }
337 40 : ret = CheckRoceAttr(channelDescFinal);
338 40 : if (ret != HCOMM_SUCCESS) {
339 0 : HCCL_ERROR("[%s] CheckRoceAttr failed, ret[%d].", __func__, ret);
340 0 : return ret;
341 : }
342 :
343 40 : channelDescFinals.push_back(channelDescFinal);
344 : }
345 40 : return HCOMM_SUCCESS;
346 : }
347 : } // namespace
348 :
349 : // 集合通信使用，待归一到HcommChannelCreate
350 22 : HcommResult HcommCollectiveChannelCreate(
351 : EndpointHandle endpointHandle, CommEngine engine, HcommChannelDesc* channelDescs, uint32_t channelNum,
352 : ChannelHandle* channels)
353 : {
354 22 : CHK_PTR_NULL(channelDescs);
355 20 : CHK_PTR_NULL(channels);
356 20 : CHK_PRT_RET(
357 : (channelNum == 0), HCCL_ERROR("[%s]Invalid channelNum, channelNum[%u]", __func__, channelNum), HCCL_E_PARA);
358 18 : std::vector<HcommChannelDesc> channelDescFinals;
359 18 : CHK_RET(static_cast<HcclResult>(NormalizeHcommChannelDescs(channelDescs, channelNum, channelDescFinals, engine)));
360 18 : auto startut = std::chrono::steady_clock::now();
361 18 : HCCL_INFO(
362 : "[%s] START. endpointHandle[0x%llx], engine[%s], channelNum[%u].", __func__, endpointHandle,
363 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), channelNum);
364 : HcommResult ret
365 18 : = ChannelProcess::CreateChannelsLoop(endpointHandle, engine, channelDescFinals.data(), channelNum, channels);
366 18 : HCCL_INFO(
367 : "[%s] END. channelNum[%u], take time [%lld]us.", __func__, channelNum,
368 : std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::steady_clock::now() - startut).count());
369 18 : return ret;
370 18 : }
371 :
372 0 : HcommResult HcommChannelUpdateMemInfo(HcommMemHandle* memHandles, uint32_t memHandleNum, ChannelHandle channelHandle)
373 : {
374 0 : CHK_PTR_NULL(memHandles);
375 0 : CHK_PRT_RET((memHandleNum == 0), HCCL_ERROR("[%s]Invalid memHandleNum, memHandleNum is 0.", __func__), HCCL_E_PARA);
376 0 : return ChannelProcess::ChannelUpdateMemInfo(memHandles, memHandleNum, channelHandle);
377 : }
378 :
379 2 : HcommResult CreatePluginChannels(
380 : hcomm::Endpoint* endpoint, HcommChannelDesc* channelDescs, uint32_t channelNum, ChannelHandle* channels)
381 : {
382 2 : auto* epHolder = dynamic_cast<hcomm::PluginEndpointHolder*>(endpoint);
383 2 : CHK_PTR_NULL(epHolder);
384 2 : const NicPluginEntry* entry = epHolder->GetPluginEntry();
385 2 : CHK_PTR_NULL(entry);
386 2 : void* epCtx = endpoint->GetNicCtx();
387 :
388 4 : for (uint32_t idx = 0; idx < channelNum; ++idx) {
389 2 : HcommResult ret = CreateOnePluginChannel(entry, epCtx, &channelDescs[idx], &channels[idx]);
390 2 : if (ret != HCCL_SUCCESS) {
391 0 : (void)RollbackPluginChannels(channels, idx);
392 0 : return ret;
393 : }
394 : }
395 :
396 2 : return HCCL_SUCCESS;
397 : }
398 :
399 27 : HcommResult HcommChannelCreate(
400 : EndpointHandle endpointHandle, CommEngine engine, HcommChannelDesc* channelDescs, uint32_t channelNum,
401 : ChannelHandle* channels)
402 : {
403 27 : CHK_PTR_NULL(endpointHandle);
404 24 : CHK_PTR_NULL(channelDescs);
405 23 : CHK_PTR_NULL(channels);
406 23 : CHK_PRT_RET(
407 : (channelNum == 0), HCCL_ERROR("[%s]Invalid channelNum, channelNum[%u]", __func__, channelNum), HCCL_E_PARA);
408 22 : std::vector<HcommChannelDesc> channelDescFinals;
409 22 : CHK_RET(static_cast<HcclResult>(NormalizeHcommChannelDescs(channelDescs, channelNum, channelDescFinals, engine)));
410 22 : auto endpoint = GetEndpointMap().GetEndpoint(endpointHandle);
411 22 : auto startut = std::chrono::steady_clock::now();
412 22 : HCCL_INFO(
413 : "[%s] START. endpointHandle[0x%llx], engine[%s], channelNum[%u].", __func__, endpointHandle,
414 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), channelNum);
415 22 : if (endpoint != nullptr && endpoint->GetNicOps() != nullptr && endpoint->GetNicOps() != &g_BuiltinEndpointOps) {
416 2 : CHK_RET(
417 : static_cast<HcclResult>(CreatePluginChannels(endpoint, channelDescFinals.data(), channelNum, channels)));
418 2 : HCCL_INFO(
419 : "[%s] END. channelNum[%u], take time [%lld]us.", __func__, channelNum,
420 : std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::steady_clock::now() - startut).count());
421 2 : return HCCL_SUCCESS;
422 : }
423 20 : (void)HcommResMgrInit();
424 20 : if (endpoint != nullptr) {
425 18 : CHK_RET(RefreshEndpointContext(endpoint->GetEndpointDesc()));
426 : }
427 20 : std::vector<ChannelHandle> hostChannelHandles(channelNum);
428 20 : ChannelHandle* targetChannels = hostChannelHandles.data();
429 20 : CHK_RET(ChannelProcess::CreateChannelsLoop(
430 : endpointHandle, engine, channelDescFinals.data(), channelNum, targetChannels));
431 20 : CHK_RET(
432 : ChannelProcess::PrepareUserChannels(targetChannels, channels, channelDescFinals.data(), channelNum, engine));
433 19 : HCCL_INFO(
434 : "[%s] END. channelNum[%u], take time [%lld]us.", __func__, channelNum,
435 : std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::steady_clock::now() - startut).count());
436 19 : HCCL_RUN_INFO(
437 : "[%s] channels created, channelNum[%u], engine[%s]", __func__, channelNum,
438 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str());
439 19 : return HCCL_SUCCESS;
440 22 : }
441 :
442 59 : HcommResult HcommChannelGet(ChannelHandle channelHandle, void** channel)
443 : {
444 59 : CHK_PTR_NULL(channel);
445 58 : return ChannelProcess::ChannelGet(channelHandle, channel);
446 : }
447 :
448 36 : HcommResult HcommChannelGetStatus(const ChannelHandle* channelList, uint32_t listNum, int32_t* statusList)
449 : {
450 36 : CHK_PTR_NULL(channelList);
451 34 : CHK_PTR_NULL(statusList);
452 32 : CHK_PRT_RET((listNum == 0), HCCL_ERROR("[%s]Invalid listNum, listNum[%u]", __func__, listNum), HCCL_E_PARA);
453 :
454 30 : if (IS_PLUGIN_HANDLE(channelList[0])) {
455 3 : for (uint32_t i = 0; i < listNum; i++) {
456 2 : auto* ch = CHANNEL_FROM_HANDLE(channelList[i]);
457 3 : CHK_PTR_NULL(ch);
458 2 : int32_t status = 0;
459 2 : HcommResult ret = static_cast<HcommResult>(ch->GetNicOps()->getStatus(ch->GetNicCtx(), &status));
460 2 : if (ret != HCCL_SUCCESS) {
461 1 : HCCL_ERROR("[%s] plugin getStatus failed, idx[%u], ret[%d].", __func__, i, ret);
462 1 : return ret;
463 : }
464 1 : statusList[i] = status;
465 : }
466 1 : return HCCL_SUCCESS;
467 : } else {
468 28 : (void)HcommResMgrInit();
469 28 : std::vector<CommEngine> engines;
470 28 : std::vector<HcommChannelDesc> channelDescFinals;
471 28 : std::vector<ChannelStatus> internalStatus(listNum);
472 28 : auto startut = std::chrono::steady_clock::now();
473 : HcclResult ret
474 28 : = ChannelProcess::GetChannelsInfo(channelList, listNum, engines, channelDescFinals, internalStatus);
475 28 : if (ret != HCCL_SUCCESS) {
476 2 : HCCL_ERROR("[%s] GetChannelsInfo failed, ret[%d]", __func__, ret);
477 2 : return HCCL_E_INTERNAL;
478 : }
479 26 : ret = ChannelProcess::HandleStatusByEngine(
480 : channelList, listNum, engines, channelDescFinals, internalStatus, statusList);
481 26 : if (ret != HCCL_SUCCESS) {
482 0 : HCCL_ERROR("[%s] HandleStatusByEngine failed, ret[%d]", __func__, ret);
483 0 : return HCCL_E_INTERNAL;
484 : }
485 26 : HCCL_INFO(
486 : "[%s] END. listNum[%u], take time [%lld]us.", __func__, listNum,
487 : std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::steady_clock::now() - startut).count());
488 26 : return HCCL_SUCCESS;
489 28 : }
490 : }
491 :
492 2 : HcommResult HcommChannelGetNotifyNum(ChannelHandle channelHandle, uint32_t* notifyNum)
493 : {
494 2 : CHK_PTR_NULL(notifyNum);
495 1 : return ChannelProcess::ChannelGetNotifyNum(channelHandle, notifyNum);
496 : }
497 :
498 19 : static HcclResult DestroyBuiltinChannels(std::vector<ChannelHandle>& builtinChannels)
499 : {
500 : // 即使 plugin channel 销毁失败，也需继续销毁 builtin channel，避免 RDMA/jetty 资源泄漏
501 : // 及 SharedJettyMgr 残留记录永久阻塞 Endpoint 销毁。最终返回首个错误（优先 plugin 错误）。
502 19 : HcclResult builtinRet = HCCL_SUCCESS;
503 19 : if (builtinChannels.empty()) {
504 0 : return builtinRet;
505 : }
506 38 : builtinRet = ChannelProcess::ChannelDestroy(
507 19 : builtinChannels.data(), builtinChannels.size(), AicpuTsChannelHelper::GetBinHandle());
508 : // 无论 ChannelDestroy 成功与否都注销 SharedJettyMgr 记录：
509 : // 成功时正常清理；失败时 channel 已不可用，若不注销会永久阻塞 Endpoint 销毁。
510 19 : if (builtinRet != HCCL_SUCCESS) {
511 0 : HCCL_WARNING(
512 : "[%s] ChannelDestroy failed, ret[%d], force unregister shared jetty channels.", __func__, builtinRet);
513 : }
514 19 : (void)hcomm::SharedJettyMgr::GetInstance().UnregisterChannels(builtinChannels.data(), builtinChannels.size());
515 19 : return builtinRet;
516 : }
517 :
518 23 : HcommResult HcommChannelDestroy(const ChannelHandle* channels, uint32_t channelNum)
519 : {
520 23 : CHK_PTR_NULL(channels);
521 22 : CHK_PRT_RET(
522 : (channelNum == 0), HCCL_ERROR("[%s]Invalid channelNum, channelNum[%u]", __func__, channelNum), HCCL_E_PARA);
523 21 : if (IS_PLUGIN_HANDLE(channels[0])) {
524 4 : for (uint32_t idx = 0; idx < channelNum; ++idx) {
525 2 : auto* ch = CHANNEL_FROM_HANDLE(channels[idx]);
526 2 : HcclResult ret = ChannelProcess::RemovePluginChannelFromMap(reinterpret_cast<ChannelHandle>(ch));
527 2 : if (ret != HCCL_SUCCESS) {
528 0 : HCCL_WARNING(
529 : "[%s] plugin channel not found in map during destroy, handle[0x%llx], ret[%d].", __func__,
530 : channels[idx], ret);
531 : }
532 : }
533 2 : return HCCL_SUCCESS;
534 : }
535 19 : (void)HcommResMgrInit();
536 19 : std::vector<ChannelHandle> builtinChannels;
537 19 : builtinChannels.reserve(channelNum);
538 39 : for (uint32_t idx = 0; idx < channelNum; ++idx) {
539 20 : builtinChannels.push_back(channels[idx]);
540 : }
541 19 : return static_cast<HcommResult>(DestroyBuiltinChannels(builtinChannels));
542 19 : }
543 :
544 0 : HcommResult HcommChannelConfigCreate(HcommChannelConfig* config)
545 : {
546 0 : return static_cast<HcommResult>(hcomm::ChannelConfigCreate(config));
547 : }
548 :
549 0 : HcommResult HcommChannelConfigDestroy(HcommChannelConfig config)
550 : {
551 0 : return static_cast<HcommResult>(hcomm::ChannelConfigDestroy(config));
552 : }
553 :
554 0 : HcommResult HcommChannelConfigSetInt(HcommChannelConfig config, HcommChannelConfigType type, uint32_t value)
555 : {
556 0 : return static_cast<HcommResult>(hcomm::ChannelConfigSetInt(config, type, value));
557 : }
558 :
559 0 : static bool IsUbProtocol(CommProtocol protocol)
560 : {
561 0 : return protocol == COMM_PROTOCOL_UB_CTP || protocol == COMM_PROTOCOL_UBC_TP;
562 : }
563 :
564 0 : static HcclResult ValidateSharedQueueConfig(const std::vector<HcommChannelDesc>& channelDescs)
565 : {
566 0 : for (uint32_t i = 0; i < channelDescs.size(); ++i) {
567 0 : CommProtocol protocol = channelDescs[i].remoteEndpoint.protocol;
568 0 : if (!IsUbProtocol(protocol)) {
569 0 : HCCL_ERROR(
570 : "[%s] IS_SHARED_QUEUE only supports UB protocols (UB_CTP/UBC_TP), "
571 : "channelDesc[%u] protocol[%d].",
572 : __func__, i, protocol);
573 0 : return HCCL_E_NOT_SUPPORT;
574 : }
575 : }
576 0 : return HCCL_SUCCESS;
577 : }
578 :
579 0 : static HcclResult CreateAndRegisterSharedQueueBuiltinChannels(
580 : EndpointHandle endpointHandle, CommEngine engine, HcommChannelDesc* channelDescFinals, uint32_t channelNum,
581 : ChannelHandle* channels)
582 : {
583 : // 共享模式建链流程与 HcommChannelCreate 一致：CreateChannelsLoop 传 isSharedQueue=true，
584 : // channel 的 BuildConnection 据此走共享 jetty 复用路径；PrepareUserChannels 完成 AICPU/AIV 预分配。
585 0 : std::vector<ChannelHandle> hostChannelHandles(channelNum);
586 0 : ChannelHandle* targetChannels = hostChannelHandles.data();
587 :
588 0 : CHK_RET(ChannelProcess::CreateChannelsLoop(
589 : endpointHandle, engine, channelDescFinals, channelNum, targetChannels, true));
590 : HcclResult prepRet
591 0 : = ChannelProcess::PrepareUserChannels(targetChannels, channels, channelDescFinals, channelNum, engine);
592 0 : if (prepRet != HCCL_SUCCESS) {
593 0 : HCCL_ERROR("[%s] PrepareUserChannels failed, ret[%d], destroying created channels.", __func__, prepRet);
594 0 : (void)ChannelProcess::ChannelDestroy(targetChannels, channelNum, AicpuTsChannelHelper::GetBinHandle());
595 0 : return prepRet;
596 : }
597 :
598 0 : HcclResult regRet = hcomm::SharedJettyMgr::GetInstance().RegisterChannels(endpointHandle, channels, channelNum);
599 0 : if (regRet != HCCL_SUCCESS) {
600 0 : HCCL_ERROR("[%s] failed to register shared jetty channels, ret[%d].", __func__, regRet);
601 0 : (void)ChannelProcess::ChannelDestroy(channels, channelNum, AicpuTsChannelHelper::GetBinHandle());
602 0 : return regRet;
603 : }
604 0 : return HCCL_SUCCESS;
605 0 : }
606 :
607 0 : HcommResult HcommChannelCreateWithConfig(
608 : EndpointHandle endpointHandle, CommEngine engine, HcommChannelDesc* channelDescs, uint32_t channelNum,
609 : HcommChannelConfig config, ChannelHandle* channels)
610 : {
611 0 : CHK_PTR_NULL(endpointHandle);
612 0 : CHK_PTR_NULL(channelDescs);
613 0 : CHK_PTR_NULL(channels);
614 0 : CHK_PRT_RET(
615 : (channelNum == 0), HCCL_ERROR("[%s]Invalid channelNum, channelNum[%u]", __func__, channelNum), HCCL_E_PARA);
616 0 : HCCL_INFO(
617 : "[%s] START. endpointHandle[0x%llx], engine[%s], channelNum[%u], config[%p].", __func__, endpointHandle,
618 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), channelNum, config);
619 :
620 0 : bool isSharedQueue = false;
621 0 : if (config != nullptr) {
622 0 : auto* cfg = static_cast<hcomm::HcommChannelConfigData*>(config);
623 0 : isSharedQueue = cfg->isSharedQueue;
624 : }
625 :
626 : // 非共享模式直接复用 HcommChannelCreate 流程，避免重复维护两套建链逻辑
627 0 : if (!isSharedQueue) {
628 0 : return HcommChannelCreate(endpointHandle, engine, channelDescs, channelNum, channels);
629 : }
630 :
631 : // 共享 jetty 仅支持 AIV 引擎：AICPU 等 channel 的 BuildConnection 不处理共享 jetty 路径，
632 : // 强行创建会导致 channel 注册到 SharedJettyMgr 但无实际 jetty 共享，多 channel 共用同一 SQ
633 : // 但 PI/CI 未协调，引发 WQE 覆盖、doorbell 不前进、notify 超时。
634 0 : if (engine != COMM_ENGINE_AIV) {
635 0 : HCCL_ERROR(
636 : "[%s] IS_SHARED_QUEUE currently only supports AIV engine, engine[%d].", __func__, static_cast<int>(engine));
637 0 : return HCCL_E_NOT_SUPPORT;
638 : }
639 :
640 0 : auto endpoint = GetEndpointMap().GetEndpoint(endpointHandle);
641 0 : if (endpoint != nullptr) {
642 0 : CHK_RET(RefreshEndpointContext(endpoint->GetEndpointDesc()));
643 : }
644 0 : (void)HcommResMgrInit();
645 :
646 0 : std::vector<HcommChannelDesc> channelDescFinals;
647 0 : CHK_RET(static_cast<HcclResult>(NormalizeHcommChannelDescs(channelDescs, channelNum, channelDescFinals, engine)));
648 : // NormalizeHcommChannelDescs 内部已调 CheckUbAttr，此处仅补共享模式专有校验
649 0 : CHK_RET(ValidateSharedQueueConfig(channelDescFinals));
650 :
651 0 : HcclResult ret = CreateAndRegisterSharedQueueBuiltinChannels(
652 : endpointHandle, engine, channelDescFinals.data(), channelNum, channels);
653 0 : if (ret != HCCL_SUCCESS) {
654 0 : return static_cast<HcommResult>(ret);
655 : }
656 :
657 0 : HCCL_INFO("[%s] SUCCESS. isSharedQueue[%d], channelNum[%u].", __func__, isSharedQueue, channelNum);
658 0 : return HCCL_SUCCESS;
659 0 : }
660 :
661 : HcommResult
662 7 : HcommChannelGetRemoteMems(ChannelHandle channelHandle, uint32_t* memNum, CommMem** remoteMem, char*** memInfos)
663 : {
664 7 : CHK_PTR_NULL(remoteMem);
665 6 : CHK_PTR_NULL(memNum);
666 5 : CHK_PTR_NULL(memInfos);
667 :
668 5 : return ChannelProcess::ChannelGetRemoteMems(channelHandle, memNum, remoteMem, memInfos);
669 : }
|
