Line data Source code
1 : /**
2 : * Copyright (c) 2025 Huawei Technologies Co., Ltd.
3 : * This program is free software, you can redistribute it and/or modify it under the terms and conditions of
4 : * CANN Open Software License Agreement Version 2.0 (the "License").
5 : * Please refer to the License for details. You may not use this file except in compliance with the License.
6 : * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,
7 : * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
8 : * See LICENSE in the root of the software repository for the full text of the License.
9 : */
10 : #include <mutex>
11 : #include <cstring>
12 : #include <memory>
13 : #include <vector>
14 :
15 : #include "hccl/hccl_res.h"
16 : #include "hcomm_res.h"
17 : #include "hcomm_res_defs.h"
18 : #include "hcomm_result_defs.h"
19 : #include "log.h"
20 : #include "hcomm_c_adpt.h"
21 : #include "hcom_common.h"
22 : #include "endpoint.h"
23 : #include "thread.h"
24 : #include "aicpu_ts_thread.h"
25 : #include "cpu_ts_thread.h"
26 : #include "aicpu_ts_urma_channel.h"
27 : #include "mem_device_pub.h"
28 : #include "channel_param.h"
29 : #include "launch_aicpu.h"
30 : #include "comm_configer.h"
31 : #include "env_config/env_config.h"
32 : #include "endpoint_map.h"
33 : #ifdef ENABLE_EXPERIMENTAL
34 : #include "nic_plugin_dispatcher.h"
35 : #endif
36 :
37 : #include "../hcomm_res_mgr.h"
38 :
39 : #include "param_check_pub.h"
40 : #include "comm_engine_utils.h"
41 :
42 : #include "exception_handler.h"
43 : #include "hcclCommDfx.h"
44 : #include "hcclCommOp.h"
45 : #include "channel_process.h"
46 : #include "aicpu_ts_channel_helper.h"
47 : #include "aiv_channel_helper.h"
48 : #include "launch_device.h"
49 : #include "endpoint_monitor.h"
50 : #include "hcomm_adapter_runtime.h"
51 : #include "adapter_rts_common.h"
52 :
53 :
54 : namespace hcomm {
55 : static std::unordered_map<ThreadHandle, std::shared_ptr<hccl::Thread>> g_ThreadMap;
56 : } // namespace hcomm
57 :
58 : using namespace hcomm;
59 : static HcommEndpointMap g_EndpointMap;
60 :
61 : namespace {
62 :
63 171 : HcclResult RefreshCurrentDeviceContext()
64 : {
65 171 : s32 deviceLogicId = 0;
66 171 : CHK_RET(hrtGetDeviceRefresh(&deviceLogicId));
67 171 : u32 devicePhyId = 0;
68 171 : CHK_RET(hrtGetDevicePhyIdByIndex(static_cast<u32>(deviceLogicId), devicePhyId, true));
69 171 : HCCL_INFO("[RefreshCurrentDeviceContext] deviceLogicId[%d], devicePhyId[%u].", deviceLogicId, devicePhyId);
70 171 : return HCCL_SUCCESS;
71 : }
72 :
73 178 : HcclResult RefreshEndpointContext(const EndpointDesc &endpointDesc)
74 : {
75 178 : if (endpointDesc.loc.locType != ENDPOINT_LOC_TYPE_DEVICE) {
76 9 : return HCCL_SUCCESS;
77 : }
78 169 : return RefreshCurrentDeviceContext();
79 : }
80 :
81 14 : HcclResult RefreshCommEngineContext(CommEngine engine)
82 : {
83 14 : if (engine != COMM_ENGINE_AICPU && engine != COMM_ENGINE_AICPU_TS) {
84 12 : return HCCL_SUCCESS;
85 : }
86 2 : return RefreshCurrentDeviceContext();
87 : }
88 : }
89 :
90 : namespace {
91 67 : HcclResult ValidateEndpointDesc(const EndpointDesc *endpoint, EndpointHandle *endpointHandle)
92 : {
93 67 : CHK_PTR_NULL(endpoint);
94 66 : CHK_PTR_NULL(endpointHandle);
95 65 : if (endpoint->loc.locType != ENDPOINT_LOC_TYPE_DEVICE && endpoint->loc.locType != ENDPOINT_LOC_TYPE_HOST) {
96 0 : HCCL_ERROR("[%s] Only support END_POINT_LOCATION_DEVICE AND END_POINT_LOCATION_HOST, but "
97 : "endpoint->loc.locType is %d",
98 : __func__,
99 : endpoint->loc.locType);
100 0 : return HCCL_E_PARA;
101 : }
102 65 : return HCCL_SUCCESS;
103 : }
104 :
105 58 : HcclResult RegisterDeviceEndpointMonitorIfNeeded(const EndpointDesc *endpoint, EndpointHandle handle)
106 : {
107 58 : if ((endpoint->loc.locType != ENDPOINT_LOC_TYPE_DEVICE) ||
108 52 : ((endpoint->protocol != COMM_PROTOCOL_UBC_CTP) && (endpoint->protocol != COMM_PROTOCOL_UBC_TP))) {
109 44 : return HCCL_SUCCESS;
110 : }
111 :
112 14 : s32 devLogicIdSigned = HcclGetThreadDeviceId();
113 14 : CHK_PRT_RET(devLogicIdSigned < 0,
114 : HCCL_ERROR("[%s] HcclGetThreadDeviceId failed, ret[%d]", __func__, devLogicIdSigned), HCCL_E_INTERNAL);
115 14 : EndpointMonitor::GetInstance(devLogicIdSigned).RegisterToEndpointMonitor(devLogicIdSigned, handle);
116 14 : return HCCL_SUCCESS;
117 : }
118 :
119 64 : HcclResult CreateBuiltinEndpoint(const EndpointDesc *endpoint, EndpointHandle *endpointHandle)
120 : {
121 64 : CHK_RET(RefreshEndpointContext(*endpoint));
122 64 : std::unique_ptr<Endpoint> endpointPtr = nullptr;
123 64 : HcclResult ret = Endpoint::CreateEndpoint(*endpoint, endpointPtr);
124 64 : if (ret != HCCL_SUCCESS) {
125 2 : HCCL_ERROR("call Endpoint::CreateEndpoint failed");
126 2 : return ret;
127 : }
128 62 : CHK_PTR_NULL(endpointPtr);
129 62 : ret = endpointPtr->Init();
130 62 : if (ret != HCCL_SUCCESS) {
131 4 : HCCL_ERROR("call endpointPtr->Init failed");
132 4 : return ret;
133 : }
134 :
135 58 : const EndpointHandle handle = reinterpret_cast<EndpointHandle>(endpointPtr.get());
136 58 : CHK_PTR_NULL(handle);
137 58 : EXCEPTION_CATCH(g_EndpointMap.AddEndpoint(handle, std::move(endpointPtr)), return HCCL_E_INTERNAL);
138 58 : *endpointHandle = handle;
139 58 : CHK_RET(RegisterDeviceEndpointMonitorIfNeeded(endpoint, handle));
140 58 : HCCL_INFO("[%s] endpointDesc.protocol [%d] and endpointDesc.loc.locType [%d] create endpointHandle [%p] done.",
141 : __func__, endpoint->protocol, endpoint->loc.locType, handle);
142 58 : return HCCL_SUCCESS;
143 64 : }
144 : }
145 :
146 35 : HcommResult CheckUbAttr(HcommChannelDesc &channelDesc)
147 : {
148 35 : if (channelDesc.remoteEndpoint.protocol != COMM_PROTOCOL_UBC_TP
149 35 : && channelDesc.remoteEndpoint.protocol != COMM_PROTOCOL_UBOE
150 35 : && channelDesc.remoteEndpoint.protocol != COMM_PROTOCOL_UBG
151 35 : && channelDesc.remoteEndpoint.protocol != COMM_PROTOCOL_UBC_CTP) {
152 19 : return HCCL_SUCCESS;
153 : }
154 :
155 : // check sqDepth
156 16 : if (channelDesc.ubAttr.sqDepth == 0xFFFFFFFF) { // 0xFFFFFFFF表示使用默认值
157 12 : HCCL_INFO("[%s] use default ubAttr.sqDepth.", __func__);
158 12 : return HCCL_SUCCESS;
159 : }
160 :
161 : // sqDepth的合理范围在[16, 256]
162 4 : if (channelDesc.ubAttr.sqDepth < 16 || channelDesc.ubAttr.sqDepth > 256) {
163 2 : HCCL_ERROR("[%s] invalid ubAttr.sqDepth[%u], should be 0 or >= 16 and <= 256.", __func__, channelDesc.ubAttr.sqDepth);
164 2 : return HCCL_E_PARA;
165 : }
166 :
167 : // channelDesc.ubAttr.sqDepth调整到2的整数次幂
168 2 : auto GetNextPowerOfTwo = [](uint32_t n) -> uint32_t {
169 2 : n--;
170 2 : n |= n >> 1;
171 2 : n |= n >> 2;
172 2 : n |= n >> 4;
173 2 : n |= n >> 8;
174 2 : n |= n >> 16;
175 2 : return n + 1;
176 : };
177 :
178 2 : channelDesc.ubAttr.sqDepth = GetNextPowerOfTwo(channelDesc.ubAttr.sqDepth);
179 :
180 2 : return HCCL_SUCCESS;
181 : }
182 :
183 28 : HcommResult CheckRoceAttr(HcommChannelDesc &channelDesc)
184 : {
185 28 : if (channelDesc.remoteEndpoint.protocol != COMM_PROTOCOL_ROCE) {
186 28 : return HCCL_SUCCESS;
187 : }
188 :
189 0 : if (channelDesc.roceAttr.queueNum == INVALID_UINT) {
190 0 : channelDesc.roceAttr.queueNum = 1;
191 0 : HCCL_INFO("[%s] set roceAttr.queueNum to 1.", __func__);
192 : }
193 :
194 0 : return HCCL_SUCCESS;
195 : }
196 :
197 : namespace {
198 28 : void ApplyHcommChannelDescV1Fields(const HcommChannelDesc &channelDesc, HcommChannelDesc &channelDescFinal)
199 : {
200 28 : if (channelDesc.header.version < HCOMM_CHANNEL_VERSION_ONE) {
201 0 : return;
202 : }
203 :
204 28 : channelDescFinal.remoteEndpoint = channelDesc.remoteEndpoint;
205 28 : channelDescFinal.notifyNum = channelDesc.notifyNum;
206 28 : channelDescFinal.exchangeAllMems = channelDesc.exchangeAllMems;
207 28 : channelDescFinal.memHandles = channelDesc.memHandles;
208 28 : channelDescFinal.memHandleNum = channelDesc.memHandleNum;
209 28 : channelDescFinal.socket = channelDesc.socket;
210 28 : channelDescFinal.role = channelDesc.role;
211 28 : channelDescFinal.port = channelDesc.port;
212 : }
213 :
214 28 : HcommResult ProcessHcommChannelDescs(const HcommChannelDesc &channelDesc, HcommChannelDesc &channelDescFinal)
215 : {
216 28 : if (channelDesc.header.size < sizeof(CommAbiHeader)) {
217 0 : HCCL_ERROR("[%s] invalid channelDesc.header.size[%u].", __func__, channelDesc.header.size);
218 0 : return HCCL_E_PARA;
219 : }
220 :
221 28 : if (channelDesc.header.magicWord != channelDescFinal.header.magicWord) {
222 0 : HCCL_ERROR("[%s] channelDesc.header.magicWord[0x%08x] is invalid, expected[0x%08x].",
223 : __func__, channelDesc.header.magicWord, channelDescFinal.header.magicWord);
224 0 : return HCCL_E_PARA;
225 : }
226 :
227 28 : const uint32_t copySize = (channelDescFinal.header.size < channelDesc.header.size ?
228 28 : channelDescFinal.header.size : channelDesc.header.size) - sizeof(CommAbiHeader);
229 28 : CHK_SAFETY_FUNC_RET(memcpy_s(reinterpret_cast<uint8_t *>(&channelDescFinal) + sizeof(CommAbiHeader), copySize,
230 : reinterpret_cast<const uint8_t *>(&channelDesc) + sizeof(CommAbiHeader), copySize));
231 28 : ApplyHcommChannelDescV1Fields(channelDesc, channelDescFinal);
232 28 : if (channelDesc.header.version > HCOMM_CHANNEL_VERSION) {
233 0 : HCCL_RUN_WARNING("The version of provided [%u] is higher than the current version[%u], "
234 : "unsupported configuration will be ignored.",
235 : channelDesc.header.version, HCOMM_CHANNEL_VERSION);
236 28 : } else if (channelDesc.header.version < HCOMM_CHANNEL_VERSION) {
237 1 : HCCL_RUN_WARNING("The version of provided [%u] is lower than the current version[%u], "
238 : "configurations supported by later versions will be ignored.",
239 : channelDesc.header.version, HCOMM_CHANNEL_VERSION);
240 : }
241 :
242 : // qos：低版本时置默认值
243 28 : if (channelDesc.header.version <= HCOMM_CHANNEL_VERSION_ONE) {
244 1 : channelDescFinal.qos = 0xFFFFFFFFU;
245 : } else {
246 27 : channelDescFinal.qos = channelDesc.qos;
247 : }
248 :
249 : // v3：channelName，低版本时置 NULL
250 28 : if (channelDesc.header.version < HCOMM_CHANNEL_VERSION) {
251 1 : channelDescFinal.channelName = nullptr;
252 : } else {
253 27 : channelDescFinal.channelName = channelDesc.channelName;
254 27 : if (channelDescFinal.channelName != nullptr &&
255 1 : reinterpret_cast<uintptr_t>(channelDescFinal.channelName) == static_cast<uintptr_t>(-1)) {
256 0 : channelDescFinal.channelName = nullptr;
257 : }
258 : }
259 :
260 28 : if (channelDescFinal.channelName != nullptr) {
261 1 : size_t nameLen = strnlen(channelDescFinal.channelName, HCOMM_CHANNEL_NAME_MAX_LEN + 1);
262 1 : if (nameLen > HCOMM_CHANNEL_NAME_MAX_LEN) {
263 0 : HCCL_ERROR("[%s] channelName too long, max len[%u].", __func__, HCOMM_CHANNEL_NAME_MAX_LEN);
264 0 : return HCCL_E_PARA;
265 : }
266 : }
267 :
268 28 : return HCOMM_SUCCESS;
269 : }
270 :
271 28 : HcommResult NormalizeHcommChannelDescs(HcommChannelDesc *channelDescs, uint32_t channelNum,
272 : std::vector<HcommChannelDesc> &channelDescFinals)
273 : {
274 28 : channelDescFinals.clear();
275 28 : channelDescFinals.reserve(channelNum);
276 56 : for (uint32_t idx = 0; idx < channelNum; ++idx) {
277 28 : HcommChannelDesc channelDescFinal{};
278 28 : HcommResult ret = HcommChannelDescInit(&channelDescFinal, 1);
279 28 : if (ret != HCOMM_SUCCESS) {
280 0 : return ret;
281 : }
282 28 : ret = ProcessHcommChannelDescs(channelDescs[idx], channelDescFinal);
283 28 : if (ret != HCOMM_SUCCESS) {
284 0 : HCCL_ERROR("[%s] failed to normalize channelDesc[%u], ret[%d].", __func__, idx, ret);
285 0 : return ret;
286 : }
287 28 : ret = CheckUbAttr(channelDescFinal);
288 28 : if (ret != HCOMM_SUCCESS) {
289 0 : HCCL_ERROR("[%s] CheckUbAttr failed, ret[%d].", __func__, ret);
290 0 : return ret;
291 : }
292 28 : ret = CheckRoceAttr(channelDescFinal);
293 28 : if (ret != HCOMM_SUCCESS) {
294 0 : HCCL_ERROR("[%s] CheckRoceAttr failed, ret[%d].", __func__, ret);
295 0 : return ret;
296 : }
297 :
298 28 : channelDescFinals.push_back(channelDescFinal);
299 : }
300 28 : return HCOMM_SUCCESS;
301 : }
302 :
303 : } // namespace
304 :
305 1264 : HcommResult HcommResMgrInit(uint32_t devPhyId)
306 : {
307 1264 : bool noDevice = false;
308 1264 : if (devPhyId == UINT32_MAX) {
309 274 : CHK_RET(ResolveRuntimeDevicePhyId(devPhyId, noDevice));
310 : }
311 :
312 : // 临时方案：触发统一平台层单例触发静态对象声明
313 : // 内部流程触发各种单例声明，保证时序
314 : EXCEPTION_HANDLE_BEGIN
315 1264 : if (noDevice) {
316 0 : (void)HcommResMgr::GetInstance(devPhyId);
317 0 : return HCCL_SUCCESS;
318 : }
319 :
320 1414 : HCCLV2_FUNC_RUN([&]() -> HcclResult {
321 : (void)HcommResMgr::GetInstance(devPhyId);
322 : return HcclResult::HCCL_SUCCESS;
323 : }());
324 0 : EXCEPTION_HANDLE_END
325 1114 : return HCCL_SUCCESS;
326 : }
327 :
328 20 : HcommResult HcommEndpointGet(EndpointHandle endpointHandle, void **endpoint) // 根据endpointHandle返回Endpoint对象指针
329 : {
330 20 : CHK_PTR_NULL(endpoint);
331 : #ifdef ENABLE_EXPERIMENTAL
332 17 : bool handled = false;
333 17 : CHK_RET(static_cast<HcclResult>(PluginEndpointGet(endpointHandle, endpoint, handled)));
334 17 : if (handled) {
335 1 : return HCCL_SUCCESS;
336 : }
337 : #endif
338 :
339 16 : auto it = g_EndpointMap.GetEndpoint(endpointHandle);
340 16 : CHK_PRT_RET(it == nullptr, HCCL_ERROR("[%s] endpoint not found, endpointHandle[%p]",
341 : __func__, endpointHandle), HCCL_E_NOT_FOUND);
342 :
343 14 : *endpoint = static_cast<void *>(it);
344 14 : HCCL_INFO("[%s] START. endpointHandle[0x%llx] endpoint[0x%llx].",__func__, endpointHandle, endpoint);
345 14 : return HCCL_SUCCESS;
346 : }
347 :
348 67 : HcommResult HcommEndpointCreate(const EndpointDesc *endpoint, EndpointHandle *endpointHandle)
349 : {
350 : EXCEPTION_HANDLE_BEGIN
351 67 : (void)HcommResMgrInit();
352 74 : CHK_RET(ValidateEndpointDesc(endpoint, endpointHandle));
353 : #ifdef ENABLE_EXPERIMENTAL
354 65 : bool pluginHandled = false;
355 65 : CHK_RET(static_cast<HcclResult>(PluginEndpointCreate(endpoint, endpointHandle, pluginHandled)));
356 65 : if (pluginHandled) {
357 1 : HCCL_INFO("[NicPluginDebug][%s] plugin endpoint created, protocol[%d], handle[%p].",
358 : __func__, endpoint->protocol, *endpointHandle);
359 1 : return HCCL_SUCCESS;
360 : }
361 : #endif
362 64 : CHK_RET(CreateBuiltinEndpoint(endpoint, endpointHandle));
363 58 : HcommResMgr::RegisterDeviceResetCallback();
364 0 : EXCEPTION_HANDLE_END
365 58 : return HCCL_SUCCESS;
366 : }
367 :
368 32 : HcommResult HcommEndpointDestroy(EndpointHandle endpointHandle)
369 : {
370 32 : (void)HcommResMgrInit();
371 32 : HCCL_INFO("[%s] START. endpointHandle[0x%llx].",__func__, endpointHandle);
372 : #ifdef ENABLE_EXPERIMENTAL
373 32 : bool handled = false;
374 32 : CHK_RET(static_cast<HcclResult>(PluginEndpointDestroy(endpointHandle, handled)));
375 32 : if (handled) {
376 1 : return HCCL_SUCCESS;
377 : }
378 : #endif
379 :
380 31 : auto endpoint = g_EndpointMap.GetEndpoint(endpointHandle);
381 31 : if (endpoint != nullptr) {
382 29 : CHK_RET(RefreshEndpointContext(endpoint->GetEndpointDesc()));
383 : }
384 31 : s32 devLogicIdSigned = HcclGetThreadDeviceId();
385 31 : CHK_PRT_RET(devLogicIdSigned < 0,
386 : HCCL_ERROR("[%s] HcclGetThreadDeviceId failed, ret[%d]", __func__, devLogicIdSigned), HCCL_E_INTERNAL);
387 31 : EndpointMonitor::GetInstance(devLogicIdSigned).RemoveEpHandleFromEndpointMonitor(endpointHandle);
388 31 : auto ret = g_EndpointMap.RemoveEndpoint(endpointHandle);
389 31 : CHK_PRT_RET(ret == false, HCCL_ERROR("[%s] endpoint not found, endpointHandle[0x%llx]",
390 : __func__, endpointHandle), HCCL_E_NOT_FOUND);
391 29 : endpointHandle = nullptr;
392 :
393 29 : return HCCL_SUCCESS;
394 : }
395 :
396 :
397 12 : HcommResult HcommEndpointStartListen(EndpointHandle endpointHandle, uint32_t port, HcommEndpointListenConfig* config)
398 : {
399 : (void)config;
400 12 : auto endpoint = g_EndpointMap.GetEndpoint(endpointHandle);
401 12 : CHK_PRT_RET(endpoint == nullptr, HCCL_ERROR("[%s] endpoint not found, endpointHandle[%p]",
402 : __func__, endpointHandle), HCCL_E_NOT_FOUND);
403 10 : CHK_RET(endpoint->ServerSocketListen(port));
404 10 : return HCCL_SUCCESS;
405 : }
406 :
407 4 : HcommResult HcommEndpointStopListen(EndpointHandle endpointHandle, uint32_t port)
408 : {
409 4 : auto endpoint = g_EndpointMap.GetEndpoint(endpointHandle);
410 4 : CHK_PRT_RET(endpoint == nullptr, HCCL_ERROR("[%s] endpoint not found, endpointHandle[%p]",
411 : __func__, endpointHandle), HCCL_E_NOT_FOUND);
412 2 : CHK_RET(endpoint->ServerSocketStopListen(port));
413 2 : return HCCL_SUCCESS;
414 : }
415 :
416 6 : HcommResult HcommEndpointGetListenPort(EndpointHandle endpointHandle, uint32_t *port)
417 : {
418 6 : CHK_PTR_NULL(port);
419 4 : (void)HcommResMgrInit();
420 : #ifdef ENABLE_EXPERIMENTAL
421 4 : if (IsPluginEndpoint(endpointHandle)) {
422 1 : return HCCL_E_NOT_SUPPORT;
423 : }
424 : #endif
425 :
426 3 : auto endpoint = g_EndpointMap.GetEndpoint(endpointHandle);
427 3 : CHK_PRT_RET(endpoint == nullptr, HCCL_ERROR("[%s] endpoint not found, endpointHandle[%p]",
428 : __func__, endpointHandle), HCCL_E_NOT_FOUND);
429 1 : CHK_RET(RefreshEndpointContext(endpoint->GetEndpointDesc()));
430 1 : return endpoint->ServerSocketGetListenPort(port);
431 : }
432 :
433 37 : HcommResult HcommMemReg(EndpointHandle endpointHandle, const char *memTag, const CommMem *mem,
434 : HcommMemHandle *memHandle)
435 : {
436 37 : CHK_PTR_NULL(memHandle);
437 : EXCEPTION_HANDLE_BEGIN
438 46 : CHK_PTR_NULL(mem);
439 34 : CHK_PTR_NULL(memHandle);
440 34 : (void)HcommResMgrInit();
441 34 : HCCL_INFO("[%s] START. endpointHandle[0x%llx].",__func__, endpointHandle);
442 : #ifdef ENABLE_EXPERIMENTAL
443 34 : bool handled = false;
444 34 : CHK_RET(static_cast<HcclResult>(PluginMemReg(endpointHandle, memTag, mem, memHandle, handled)));
445 34 : if (handled) {
446 1 : return HCCL_SUCCESS;
447 : }
448 : #endif
449 :
450 33 : auto endpoint = g_EndpointMap.GetEndpoint(endpointHandle);
451 33 : CHK_PRT_RET(endpoint == nullptr, HCCL_ERROR("[%s] endpoint not found, endpointHandle[0x%llx]",
452 : __func__, endpointHandle), HCCL_E_NOT_FOUND);
453 32 : CHK_RET(RefreshEndpointContext(endpoint->GetEndpointDesc()));
454 32 : CHK_RET(endpoint->RegisterMemory(*mem, memTag, reinterpret_cast<void **>(memHandle)));
455 0 : EXCEPTION_HANDLE_END
456 23 : return HCCL_SUCCESS;
457 : }
458 :
459 41 : HcommResult HcommMemUnreg(EndpointHandle endpointHandle, HcommMemHandle memHandle)
460 : {
461 41 : CHK_PTR_NULL(memHandle);
462 34 : (void)HcommResMgrInit();
463 : EXCEPTION_HANDLE_BEGIN
464 34 : HCCL_INFO("[%s] START. endpointHandle[0x%llx].",__func__, endpointHandle);
465 : #ifdef ENABLE_EXPERIMENTAL
466 34 : bool handled = false;
467 45 : CHK_RET(static_cast<HcclResult>(PluginMemUnreg(endpointHandle, memHandle, handled)));
468 34 : if (handled) {
469 1 : return HCCL_SUCCESS;
470 : }
471 : #endif
472 :
473 33 : auto endpoint = g_EndpointMap.GetEndpoint(endpointHandle);
474 33 : CHK_PRT_RET(endpoint == nullptr, HCCL_ERROR("[%s] endpoint not found, endpointHandle[0x%llx]",
475 : __func__, endpointHandle), HCCL_E_NOT_FOUND);
476 32 : CHK_RET(RefreshEndpointContext(endpoint->GetEndpointDesc()));
477 32 : CHK_RET(endpoint->UnregisterMemory(memHandle));
478 0 : EXCEPTION_HANDLE_END
479 23 : return HCCL_SUCCESS;
480 : }
481 :
482 7 : HcommResult HcommMemExport(EndpointHandle endpointHandle, HcommMemHandle memHandle, void **memDesc,
483 : uint32_t *memDescLen)
484 : {
485 7 : CHK_PTR_NULL(memHandle);
486 6 : CHK_PTR_NULL(memDesc);
487 5 : CHK_PTR_NULL(memDescLen);
488 5 : (void)HcommResMgrInit();
489 5 : HCCL_INFO("[%s] START. endpointHandle[0x%llx].",__func__, endpointHandle);
490 : #ifdef ENABLE_EXPERIMENTAL
491 5 : bool handled = false;
492 5 : CHK_RET(static_cast<HcclResult>(PluginMemExport(endpointHandle, memHandle, memDesc, memDescLen, handled)));
493 4 : if (handled) {
494 1 : return HCCL_SUCCESS;
495 : }
496 : #endif
497 :
498 3 : auto endpoint = g_EndpointMap.GetEndpoint(endpointHandle);
499 3 : CHK_PRT_RET(endpoint == nullptr, HCCL_ERROR("[%s] endpoint not found, endpointHandle[0x%llx]",
500 : __func__, endpointHandle), HCCL_E_NOT_FOUND);
501 2 : CHK_RET(RefreshEndpointContext(endpoint->GetEndpointDesc()));
502 2 : CHK_RET(endpoint->MemoryExport(memHandle, memDesc, memDescLen));
503 2 : return HCCL_SUCCESS;
504 : }
505 :
506 5 : HcommResult HcommMemImport(EndpointHandle endpointHandle, const void *memDesc, uint32_t descLen, CommMem *outMem)
507 : {
508 5 : CHK_PTR_NULL(memDesc);
509 4 : CHK_PTR_NULL(outMem);
510 4 : CHK_PRT_RET(descLen == 0, HCCL_ERROR("[%s] descLen[0] is invalid", __func__), HCCL_E_PARA);
511 4 : (void)HcommResMgrInit();
512 4 : HCCL_INFO("[%s] START. endpointHandle[0x%llx].",__func__, endpointHandle);
513 : #ifdef ENABLE_EXPERIMENTAL
514 4 : bool handled = false;
515 4 : CHK_RET(static_cast<HcclResult>(PluginMemImport(endpointHandle, memDesc, descLen, outMem, handled)));
516 3 : if (handled) {
517 1 : return HCCL_SUCCESS;
518 : }
519 : #endif
520 :
521 2 : auto endpoint = g_EndpointMap.GetEndpoint(endpointHandle);
522 2 : CHK_PRT_RET(endpoint == nullptr, HCCL_ERROR("[%s] endpoint not found, endpointHandle[0x%llx]",
523 : __func__, endpointHandle), HCCL_E_NOT_FOUND);
524 2 : CHK_RET(RefreshEndpointContext(endpoint->GetEndpointDesc()));
525 2 : CHK_PTR_NULL(outMem);
526 2 : CommMem importedMem{};
527 2 : CHK_RET(endpoint->MemoryImport(memDesc, descLen, &importedMem));
528 2 : *outMem = importedMem;
529 2 : return HCCL_SUCCESS;
530 : }
531 :
532 5 : HcommResult HcommMemUnimport(EndpointHandle endpointHandle, const void *memDesc, uint32_t descLen)
533 : {
534 5 : CHK_PTR_NULL(memDesc);
535 4 : (void)HcommResMgrInit();
536 4 : HCCL_INFO("[%s] START. endpointHandle[0x%llx].",__func__, endpointHandle);
537 : #ifdef ENABLE_EXPERIMENTAL
538 4 : bool handled = false;
539 4 : CHK_RET(static_cast<HcclResult>(PluginMemUnimport(endpointHandle, memDesc, descLen, handled)));
540 3 : if (handled) {
541 1 : return HCCL_SUCCESS;
542 : }
543 : #endif
544 :
545 2 : auto endpoint = g_EndpointMap.GetEndpoint(endpointHandle);
546 2 : CHK_PRT_RET(endpoint == nullptr, HCCL_ERROR("[%s] endpoint not found, endpointHandle[0x%llx]",
547 : __func__, endpointHandle), HCCL_E_NOT_FOUND);
548 2 : CHK_RET(RefreshEndpointContext(endpoint->GetEndpointDesc()));
549 2 : CHK_RET(endpoint->MemoryUnimport(memDesc, descLen));
550 2 : return HCCL_SUCCESS;
551 : }
552 :
553 : /* 暂未实现 */
554 2 : HcommResult HcommMemGrant(EndpointHandle endpointHandle, const HcommMemGrantInfo *remoteGrantInfo)
555 : {
556 2 : CHK_PTR_NULL(remoteGrantInfo);
557 2 : HCCL_INFO("[%s] START. endpointHandle[0x%llx].",__func__, endpointHandle);
558 :
559 2 : auto endpoint = g_EndpointMap.GetEndpoint(endpointHandle);
560 2 : CHK_PRT_RET(endpoint == nullptr, HCCL_ERROR("[%s] endpoint not found, endpointHandle[0x%llx]",
561 : __func__, endpointHandle), HCCL_E_NOT_FOUND);
562 0 : CHK_RET(endpoint->MemoryGrant(remoteGrantInfo));
563 0 : return HCCL_SUCCESS;
564 : }
565 :
566 : /* 暂未实现 */
567 0 : HcommResult HcommMemRemap(const EndpointHandle endpointHandle, const CommMem *memArray, uint64_t arraySize)
568 : {
569 0 : return HCCL_E_NOT_SUPPORT;
570 : }
571 :
572 3 : HcommResult HcommMemGetAllMemHandles(EndpointHandle endpointHandle, void **memHandles, uint32_t *memHandleNum)
573 : {
574 3 : CHK_PTR_NULL(memHandles);
575 2 : CHK_PTR_NULL(memHandleNum);
576 :
577 2 : auto endpoint = g_EndpointMap.GetEndpoint(endpointHandle);
578 2 : CHK_PRT_RET(endpoint == nullptr, HCCL_ERROR("[%s] endpoint not found, endpointHandle[0x%llx]",
579 : __func__, endpointHandle), HCCL_E_NOT_FOUND);
580 0 : CHK_RET(endpoint->GetAllMemHandles(memHandles, memHandleNum));
581 0 : return HCCL_SUCCESS;
582 : }
583 :
584 : // 集合通信使用，待归一到HcommChannelCreate
585 12 : HcommResult HcommCollectiveChannelCreate(EndpointHandle endpointHandle, CommEngine engine,
586 : HcommChannelDesc *channelDescs, uint32_t channelNum, ChannelHandle *channels)
587 : {
588 12 : CHK_PTR_NULL(channelDescs);
589 10 : CHK_PTR_NULL(channels);
590 10 : CHK_PRT_RET((channelNum == 0), HCCL_ERROR("[%s]Invalid channelNum, channelNum[%u]",
591 : __func__, channelNum), HCCL_E_PARA);
592 8 : HCCL_INFO("[%s] START. endpointHandle[0x%llx], engine[%s], channelNum[%u].",
593 : __func__, endpointHandle, GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), channelNum);
594 :
595 8 : std::vector<HcommChannelDesc> channelDescFinals;
596 8 : CHK_RET(static_cast<HcclResult>(NormalizeHcommChannelDescs(channelDescs, channelNum, channelDescFinals)));
597 8 : return ChannelProcess::CreateChannelsLoop(endpointHandle, engine, channelDescFinals.data(), channelNum, channels);
598 8 : }
599 :
600 0 : HcommResult HcommChannelUpdateMemInfo(HcommMemHandle *memHandles, uint32_t memHandleNum, ChannelHandle channelHandle)
601 : {
602 0 : CHK_PTR_NULL(memHandles);
603 0 : CHK_PRT_RET((memHandleNum == 0), HCCL_ERROR("[%s]Invalid memHandleNum, memHandleNum is 0.", __func__),
604 : HCCL_E_PARA);
605 : #ifdef ENABLE_EXPERIMENTAL
606 0 : bool handled = false;
607 0 : CHK_RET(static_cast<HcclResult>(PluginChannelUpdateMemInfo(channelHandle, memHandles, memHandleNum, handled)));
608 0 : if (handled) {
609 0 : return HCCL_SUCCESS;
610 : }
611 : #endif
612 :
613 0 : return ChannelProcess::ChannelUpdateMemInfo(memHandles, memHandleNum, channelHandle);
614 : }
615 :
616 25 : HcommResult HcommChannelCreate(EndpointHandle endpointHandle, CommEngine engine,
617 : HcommChannelDesc *channelDescs, uint32_t channelNum, ChannelHandle *channels)
618 : {
619 25 : CHK_PTR_NULL(endpointHandle);
620 22 : CHK_PTR_NULL(channelDescs);
621 21 : CHK_PTR_NULL(channels);
622 21 : CHK_PRT_RET((channelNum == 0), HCCL_ERROR("[%s]Invalid channelNum, channelNum[%u]",
623 : __func__, channelNum), HCCL_E_PARA);
624 20 : HCCL_INFO("[%s] START. endpointHandle[0x%llx], engine[%s], channelNum[%u].",
625 : __func__, endpointHandle, GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), channelNum);
626 20 : auto endpoint = g_EndpointMap.GetEndpoint(endpointHandle);
627 20 : if (endpoint != nullptr) {
628 14 : CHK_RET(RefreshEndpointContext(endpoint->GetEndpointDesc()));
629 : }
630 20 : (void)HcommResMgrInit();
631 20 : std::vector<HcommChannelDesc> channelDescFinals;
632 20 : CHK_RET(static_cast<HcclResult>(NormalizeHcommChannelDescs(channelDescs, channelNum, channelDescFinals)));
633 :
634 : #ifdef ENABLE_EXPERIMENTAL
635 20 : bool pluginHandled = false;
636 20 : CHK_RET(static_cast<HcclResult>(PluginChannelCreate(endpointHandle, engine,
637 : channelDescFinals.data(), channelNum, channels, pluginHandled)));
638 17 : if (pluginHandled) {
639 1 : return HCCL_SUCCESS;
640 : }
641 : #endif
642 :
643 16 : std::vector<ChannelHandle> hostChannelHandles(channelNum);
644 16 : ChannelHandle* targetChannels = hostChannelHandles.data();
645 :
646 16 : CHK_RET(ChannelProcess::CreateChannelsLoop(endpointHandle, engine, channelDescFinals.data(), channelNum,
647 : targetChannels));
648 16 : CHK_RET(ChannelProcess::PrepareUserChannels(targetChannels, channels, channelDescFinals.data(), channelNum, engine));
649 :
650 15 : return HCCL_SUCCESS;
651 20 : }
652 :
653 43 : HcommResult HcommChannelGet(ChannelHandle channelHandle, void **channel)
654 : {
655 43 : CHK_PTR_NULL(channel);
656 : #ifdef ENABLE_EXPERIMENTAL
657 42 : bool handled = false;
658 42 : CHK_RET(static_cast<HcclResult>(PluginChannelGet(channelHandle, channel, handled)));
659 42 : if (handled) {
660 0 : return HCCL_SUCCESS;
661 : }
662 : #endif
663 42 : return ChannelProcess::ChannelGet(channelHandle, channel);
664 : }
665 :
666 30 : HcommResult HcommChannelGetStatus(const ChannelHandle *channelList, uint32_t listNum, int32_t* statusList)
667 : {
668 30 : CHK_PTR_NULL(channelList);
669 28 : CHK_PTR_NULL(statusList);
670 26 : CHK_PRT_RET((listNum == 0), HCCL_ERROR("[%s]Invalid listNum, listNum[%u]",
671 : __func__, listNum), HCCL_E_PARA);
672 24 : (void)HcommResMgrInit();
673 : #ifdef ENABLE_EXPERIMENTAL
674 24 : bool allHandled = true;
675 52 : for (uint32_t i = 0; i < listNum; i++) {
676 28 : bool handled = false;
677 28 : CHK_RET(static_cast<HcclResult>(PluginChannelGetStatus(channelList[i], &statusList[i], handled)));
678 28 : if (!handled) {
679 28 : allHandled = false;
680 : }
681 : }
682 24 : if (allHandled) {
683 0 : return HCCL_SUCCESS;
684 : }
685 : #endif
686 :
687 24 : std::vector<CommEngine> engines;
688 24 : std::vector<HcommChannelDesc> channelDescFinals;
689 24 : std::vector<ChannelStatus> internalStatus(listNum);
690 24 : HcclResult ret = ChannelProcess::GetChannelsInfo(channelList, listNum, engines, channelDescFinals, internalStatus);
691 24 : if (ret != HCCL_SUCCESS) {
692 2 : HCCL_ERROR("[%s] GetChannelsInfo failed, ret[%d]", __func__, ret);
693 2 : return HCCL_E_INTERNAL;
694 : }
695 22 : ret = ChannelProcess::HandleStatusByEngine(channelList, listNum, engines,
696 : channelDescFinals, internalStatus, statusList);
697 22 : if (ret != HCCL_SUCCESS) {
698 0 : HCCL_ERROR("[%s] HandleStatusByEngine failed, ret[%d]", __func__, ret);
699 0 : return HCCL_E_INTERNAL;
700 : }
701 22 : return HCCL_SUCCESS;
702 24 : }
703 :
704 2 : HcommResult HcommChannelGetNotifyNum(ChannelHandle channelHandle, uint32_t *notifyNum)
705 : {
706 2 : CHK_PTR_NULL(notifyNum);
707 : #ifdef ENABLE_EXPERIMENTAL
708 1 : bool handled = false;
709 1 : CHK_RET(static_cast<HcclResult>(PluginChannelGetNotifyNum(channelHandle, notifyNum, handled)));
710 1 : if (handled) {
711 0 : return HCCL_SUCCESS;
712 : }
713 : #endif
714 1 : return ChannelProcess::ChannelGetNotifyNum(channelHandle, notifyNum);
715 : }
716 :
717 18 : HcommResult HcommChannelDestroy(const ChannelHandle *channels, uint32_t channelNum)
718 : {
719 18 : CHK_PTR_NULL(channels);
720 17 : (void)HcommResMgrInit();
721 17 : CHK_PRT_RET((channelNum == 0), HCCL_ERROR("[%s]Invalid channelNum, channelNum[%u]",
722 : __func__, channelNum), HCCL_E_PARA);
723 16 : std::vector<ChannelHandle> builtinChannels;
724 16 : builtinChannels.reserve(channelNum);
725 33 : for (uint32_t idx = 0; idx < channelNum; ++idx) {
726 : #ifdef ENABLE_EXPERIMENTAL
727 17 : bool handled = false;
728 17 : CHK_RET(static_cast<HcclResult>(PluginChannelDestroy(channels[idx], handled)));
729 17 : if (handled) {
730 1 : continue;
731 : }
732 : #endif
733 16 : builtinChannels.push_back(channels[idx]);
734 : }
735 16 : if (builtinChannels.empty()) {
736 1 : return HCCL_SUCCESS;
737 : }
738 15 : return ChannelProcess::ChannelDestroy(builtinChannels.data(), builtinChannels.size(), AicpuTsChannelHelper::GetBinHandle());
739 16 : }
740 :
741 8 : HcommResult HcommChannelGetRemoteMems(ChannelHandle channelHandle, uint32_t *memNum, CommMem **remoteMem, char ***memInfos)
742 : {
743 8 : CHK_PTR_NULL(remoteMem);
744 7 : CHK_PTR_NULL(memNum);
745 6 : CHK_PTR_NULL(memInfos);
746 : #ifdef ENABLE_EXPERIMENTAL
747 6 : bool handled = false;
748 6 : CHK_RET(static_cast<HcclResult>(PluginChannelGetRemoteMems(channelHandle, memNum, remoteMem, memInfos, handled)));
749 6 : if (handled) {
750 1 : return HCCL_SUCCESS;
751 : }
752 : #endif
753 :
754 5 : return ChannelProcess::ChannelGetRemoteMems(channelHandle, memNum, remoteMem, memInfos);
755 : }
756 :
757 15 : HcommResult HcommThreadAlloc(CommEngine engine, uint32_t threadNum, const uint32_t *notifyNumPerThread,
758 : ThreadHandle *threads) {
759 15 : CHK_PTR_NULL(threads);
760 14 : CHK_PTR_NULL(notifyNumPerThread);
761 14 : (void)HcommResMgrInit();
762 14 : const uint32_t notifyNum = notifyNumPerThread[0];
763 14 : if (threadNum > 1U) {
764 7 : HCCL_RUN_WARNING("[%s] only notifyNumPerThread[0] is used currently, threadNum[%u], notifyNum[0][%u].",
765 : __func__, threadNum, notifyNum);
766 : }
767 14 : HCCL_INFO("[%s] ThreadAcquire begin. engine[%s], threadNum[%u], notifyPerThread[%u], threads[%p]",
768 : __func__, GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), threadNum, notifyNum, threads);
769 14 : CHK_RET(RefreshCommEngineContext(engine));
770 :
771 : // 1. 参数校验
772 14 : CHK_RET(hccl::ValidateThreadParams(threadNum, notifyNum));
773 :
774 : // 2. 获取引擎对应的类型
775 : hccl::NotifyLoadType notifyLoadType;
776 : hccl::StreamType streamType;
777 11 : CHK_RET(hccl::CommEngineToNotifyLoadType(engine, notifyLoadType));
778 10 : CHK_RET(hccl::CommEngineToStreamType(engine, streamType));
779 :
780 : // 3. 创建线程
781 10 : std::vector<std::shared_ptr<hccl::Thread>> newThreads;
782 10 : hccl::ThreadCreateParams params(engine, threadNum, notifyNum, notifyLoadType, streamType);
783 10 : CHK_RET(hccl::CreateAndInitThreads(params, newThreads));
784 :
785 : // 4. 插入全局映射表
786 8 : CHK_RET(hccl::SaveThreads(newThreads));
787 :
788 : // 5. 储存线程句柄
789 8 : CHK_RET(AicpuTsChannelHelper::EnsureKernelBinLoaded(engine));
790 8 : CHK_RET(hccl::StoreThreadHandles(newThreads, threads, engine, AicpuTsChannelHelper::GetBinHandle()));
791 :
792 8 : HCCL_INFO("[HcommThreadAlloc] ThreadAcquire done: engine[%s] threadNum[%u], notifyPerThread[%u]",
793 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), threadNum, notifyNum);
794 8 : return HCCL_SUCCESS;
795 10 : }
796 :
797 10 : HcommResult HcommThreadAlloc(CommEngine engine, uint32_t threadNum, uint32_t notifyNumPerThread,
798 : ThreadHandle *threads)
799 : {
800 10 : return ::HcommThreadAlloc(engine, threadNum, ¬ifyNumPerThread, threads);
801 : }
802 :
803 0 : HcommResult HcommThreadAllocWithConfig(CommEngine engine, uint32_t threadNum,
804 : ThreadType type, const ThreadConfig *config, ThreadHandle *threads)
805 : {
806 0 : CHK_PTR_NULL(threads);
807 0 : CHK_PTR_NULL(config);
808 0 : CHK_PRT_RET(type == THREAD_TYPE_INVALID, HCCL_ERROR("[%s] thread type[%d] is invalid",
809 : __func__, static_cast<int32_t>(type)), (HcommResult)HCCL_E_PARA);
810 0 : CHK_PRT_RET(engine == COMM_ENGINE_AICPU_TS || engine == COMM_ENGINE_CPU_TS,
811 : HCCL_ERROR("[%s] commEngine[%d] CPU_TS/AICPU_TS not supported, use engine with ThreadType instead",
812 : __func__, static_cast<int32_t>(engine)), (HcommResult)HCCL_E_PARA);
813 0 : CHK_PRT_RET(engine == COMM_ENGINE_AIV || engine == COMM_ENGINE_CCU,
814 : HCCL_ERROR("[%s] commEngine[%d] AIV/CCU not supported, supported engines: CPU/AICPU",
815 : __func__, static_cast<int32_t>(engine)), (HcommResult)HCCL_E_PARA);
816 0 : CHK_PRT_RET(threadNum == 0,
817 : HCCL_ERROR("[%s] threadNum[%u] is invalid", __func__, threadNum), (HcommResult)HCCL_E_PARA);
818 0 : HcommResult hcommRet = HcommResMgrInit();
819 0 : CHK_PRT_RET(hcommRet != HCCL_SUCCESS,
820 : HCCL_ERROR("[%s] HcommResMgrInit failed, ret[%d]", __func__, static_cast<int32_t>(hcommRet)), hcommRet);
821 0 : CHK_RET(RefreshCommEngineContext(engine));
822 :
823 0 : HCCL_INFO("[%s] begin. engine[%d], threadType[%d], threadNum[%u], threads[%p]",
824 : __func__, engine, static_cast<int32_t>(type), threadNum, threads);
825 :
826 : hccl::NotifyLoadType notifyLoadType;
827 : hccl::StreamType streamType;
828 0 : CHK_RET(hccl::GetNotifyLoadType(engine, type, notifyLoadType));
829 0 : CHK_RET(hccl::GetStreamType(engine, type, streamType));
830 :
831 0 : std::vector<std::shared_ptr<hccl::Thread>> newThreads;
832 0 : newThreads.reserve(threadNum);
833 0 : for (uint32_t i = 0; i < threadNum; ++i) {
834 0 : CHK_PRT_RET(config[i].header.magicWord != HCOMM_THREAD_CONFIG_MAGIC_WORD,
835 : HCCL_ERROR("[%s] config[%u] magicWord[0x%x] mismatch, expected[0x%x], call ThreadConfigInit first",
836 : __func__, i, config[i].header.magicWord, HCOMM_THREAD_CONFIG_MAGIC_WORD), (HcommResult)HCCL_E_PARA);
837 0 : CHK_RET(hccl::ValidateThreadParams(1, config[i].notifyNumPerThread));
838 0 : std::shared_ptr<hccl::Thread> threadPtr;
839 0 : HcclResult ret = hccl::CreateThread(engine, streamType, config[i].notifyNumPerThread, notifyLoadType, threadPtr);
840 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
841 : HCCL_ERROR("[%s] Failed to create thread at index[%u], ret[%d]", __func__, i, ret), (HcommResult)ret);
842 0 : ret = threadPtr->Init();
843 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
844 : HCCL_ERROR("[%s] Failed to init thread at index[%u], ret[%d]", __func__, i, ret), (HcommResult)ret);
845 0 : newThreads.emplace_back(std::move(threadPtr));
846 0 : }
847 :
848 0 : CHK_RET(hccl::SaveThreads(newThreads));
849 0 : CHK_RET(AicpuTsChannelHelper::EnsureKernelBinLoaded(engine));
850 0 : CHK_RET(hccl::StoreThreadHandles(newThreads, threads, engine, AicpuTsChannelHelper::GetBinHandle()));
851 :
852 0 : HCCL_INFO("[%s] done: engine[%d] threadType[%d] threadNum[%u]",
853 : __func__, engine, static_cast<int32_t>(type), threadNum);
854 0 : return HCCL_SUCCESS;
855 0 : }
856 :
857 9 : HcommResult HcommThreadFree(const ThreadHandle *threads, uint32_t threadNum)
858 : {
859 9 : CHK_PTR_NULL(threads);
860 8 : (void)HcommResMgrInit();
861 8 : return hccl::FreeThreads(threads, threadNum, AicpuTsChannelHelper::GetBinHandle());
862 : }
863 :
864 7 : HcommResult HcommThreadAllocWithStream(CommEngine engine,
865 : rtStream_t stream, uint32_t notifyNum, ThreadHandle *thread)
866 : {
867 7 : CHK_PTR_NULL(thread);
868 : hccl::NotifyLoadType notifyLoadType;
869 6 : CHK_RET(CommHostEngineToNotifyLoadType(engine, notifyLoadType));
870 5 : std::shared_ptr<hccl::Thread> handle;
871 5 : EXCEPTION_CATCH(handle = std::make_shared<hccl::CpuTsThread>(stream, notifyNum, notifyLoadType), return HCCL_E_PTR);
872 5 : CHK_RET(handle->Init());
873 :
874 : // 返回第一个句柄
875 4 : *thread = reinterpret_cast<ThreadHandle>(handle.get());
876 4 : hcomm::g_ThreadMap.emplace(*thread , handle);
877 :
878 4 : HCCL_INFO("[ThreadMgr] ThreadAcquireWithStream done: engine[%s] stream[%p],"
879 : "notifyNum[%u]", GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), stream, notifyNum);
880 4 : return HCCL_SUCCESS;
881 5 : }
882 :
883 11 : HcommResult HcommEngineCtxCreate(CommEngine engine, uint64_t size, void **ctx)
884 : {
885 11 : CHK_PTR_NULL(ctx);
886 10 : if (engine == COMM_ENGINE_CPU || engine == COMM_ENGINE_CPU_TS
887 3 : || engine == COMM_ENGINE_CCU) {
888 7 : *ctx = malloc(size);
889 7 : CHK_PTR_NULL(*ctx);
890 7 : auto ret = memset_s(*ctx, size, 0, size);
891 7 : if (ret != EOK) {
892 0 : HCCL_ERROR("[%s] memset_s failed, ret[%d]", __func__, ret);
893 0 : free(*ctx);
894 0 : *ctx = nullptr;
895 0 : return HCCL_E_INTERNAL;
896 : }
897 10 : } else if (engine == COMM_ENGINE_AICPU || engine == COMM_ENGINE_AICPU_TS
898 1 : || engine == COMM_ENGINE_AIV) {
899 2 : CHK_RET(hrtMalloc(ctx, size));
900 2 : } else {
901 1 : HCCL_ERROR("[%s] not support engine type[%s]", __func__, GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str());
902 1 : return HCCL_E_PARA;
903 : }
904 9 : return HCCL_SUCCESS;
905 : }
906 :
907 12 : HcommResult HcommEngineCtxDestroy(CommEngine engine, void *ctx)
908 : {
909 12 : CHK_PTR_NULL(ctx);
910 11 : if (engine == COMM_ENGINE_CPU || engine == COMM_ENGINE_CPU_TS
911 3 : || engine == COMM_ENGINE_CCU) {
912 8 : free(ctx);
913 3 : } else if (engine == COMM_ENGINE_AICPU || engine == COMM_ENGINE_AICPU_TS
914 1 : || engine == COMM_ENGINE_AIV) {
915 2 : CHK_RET(hrtFree(ctx));
916 2 : } else {
917 1 : HCCL_ERROR("[%s] invalid engine[%s]", __func__, GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str());
918 1 : return HCCL_E_PARA;
919 : }
920 10 : return HCCL_SUCCESS;
921 : }
922 :
923 6 : HcommResult HcommEngineCtxCopy(CommEngine engine, void *dstCtx, const void *srcCtx, uint64_t size)
924 : {
925 6 : CHK_PTR_NULL(dstCtx);
926 5 : CHK_PTR_NULL(srcCtx);
927 4 : if (engine == COMM_ENGINE_AICPU_TS || engine == COMM_ENGINE_AICPU
928 4 : || engine == COMM_ENGINE_AIV) {
929 : // 从Host内存拷贝到Device Context内存上
930 0 : CHK_RET(hrtMemSyncCopy(reinterpret_cast<uint8_t*>(dstCtx), size, srcCtx, size,
931 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE));
932 4 : } else if (engine == COMM_ENGINE_CPU || engine == COMM_ENGINE_CPU_TS
933 1 : || engine == COMM_ENGINE_CCU) {
934 4 : CHK_SAFETY_FUNC_RET(memcpy_s(reinterpret_cast<uint8_t*>(dstCtx), size, srcCtx, size));
935 4 : } else {
936 0 : HCCL_ERROR("[%s]copy engine ctx failed, Unsupported engine[%s]", __func__, GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str());
937 0 : return HCCL_E_PARA;
938 : }
939 4 : HCCL_INFO("[%s]copy engine ctx success, engine[%s]", __func__, GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str());
940 4 : return HCCL_SUCCESS;
941 : }
942 :
943 0 : HcommResult HcommDfxKernelLaunch(const std::string &commTag, aclrtBinHandle binHandle, HcclDfxOpInfo dfxOpInfo)
944 : {
945 : // 申请device侧内存
946 0 : hccl::DeviceMem devicePackBuf = hccl::DeviceMem::alloc(sizeof(dfxOpInfo));
947 0 : CHK_PTR_NULL(devicePackBuf.ptr());
948 :
949 : // 将dfxOpInfo信息传递给device侧
950 0 : CHK_RET(hrtMemSyncCopy(devicePackBuf.ptr(),
951 : sizeof(dfxOpInfo),
952 : &dfxOpInfo,
953 : sizeof(dfxOpInfo),
954 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE));
955 :
956 : // 创建局部流
957 0 : hccl::Stream localStream(hccl::StreamType::STREAM_TYPE_ONLINE);
958 0 : constexpr u32 aicpuStreamMode = 1;
959 0 : CHK_RET(hrtStreamSetMode(localStream.ptr(), aicpuStreamMode));
960 :
961 : // 下kernel
962 0 : std::string kernelName = "RunAicpuDfxOpInfoInitV2";
963 :
964 : struct InitTask {
965 : u64 context;
966 : char commTag[256];
967 : };
968 :
969 0 : InitTask customInitTask = {0, ""};
970 0 : customInitTask.context = reinterpret_cast<u64>(devicePackBuf.ptr());
971 0 : s32 sRet = strncpy_s(customInitTask.commTag, TAG_MAX_LENGTH, commTag.c_str(), TAG_MAX_LENGTH - 1);
972 0 : CHK_PRT_RET(sRet != EOK, HCCL_ERROR("[%s] str copy fail. return[%d]", __func__, sRet), HCCL_E_INTERNAL);
973 :
974 0 : CHK_RET(hccl::AicpuAclKernelLaunch(localStream.ptr(),
975 : reinterpret_cast<void *>(&customInitTask),
976 : sizeof(customInitTask),
977 : binHandle,
978 : kernelName,
979 : true,
980 : NOTIFY_DEFAULT_WAIT_TIME));
981 :
982 0 : CHK_RET(
983 : hcclStreamSynchronize(localStream.ptr(), hccl::CommConfiger::GetInstance().GetCommConfigExecTimeOut(commTag)));
984 :
985 0 : HCCL_INFO("[%s] channel kernel launch success.", __func__);
986 :
987 0 : return HCCL_SUCCESS;
988 0 : }
989 :
990 1 : HcommResult HcommEndpointCheckFeature(HcommEndpointFeatureType featureType, const EndpointDesc *endpointDesc, bool *value)
991 : {
992 1 : CHK_PTR_NULL(endpointDesc);
993 1 : CHK_PTR_NULL(value);
994 1 : (void)HcommResMgrInit();
995 :
996 1 : return static_cast<HcommResult>(Endpoint::CheckFeature(*endpointDesc, featureType, *value));
997 : }
|
