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