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