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