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 :
11 : #include <atomic>
12 : #include <unordered_map>
13 : #include <mutex>
14 : #include <memory>
15 : #include <vector>
16 : #include <string>
17 : #include "hccl/hccl_res.h"
18 : #include "stream_pub.h"
19 : #include "hccl_comm_pub.h"
20 : #include "hccl_independent_common.h"
21 : #include "coll_comm_profiling.h"
22 : #include "comm_engine_utils.h"
23 : #include "coll_comm_mgr.h"
24 : #include "orion_adapter_rts.h"
25 : #include "hccl_common.h"
26 : #include "adapter_rts.h"
27 : #include "hcclCommOp.h"
28 : using namespace hccl;
29 : constexpr u32 MAX_EXPORT_THREAD_NUM = 40U;
30 : static const std::unordered_set<HcclDedicatedThreadType> ORDER_LAUNCH_TYPES = {
31 : HCCL_DED_THREAD_TYPE_AICPU_ORDER_LAUNCH_OPBASE,
32 : HCCL_DED_THREAD_TYPE_AICPU_ORDER_LAUNCH_ACLGRAPH,
33 : HCCL_DED_THREAD_TYPE_AICPU_ORDER_LAUNCH_GE,
34 : HCCL_DED_THREAD_TYPE_AICPU_ORDER_LAUNCH_DEVICE,
35 : };
36 :
37 11 : HcclResult HcclGetNotifyNumInThread(HcclComm comm, ThreadHandle thread, CommEngine engine, uint32_t* notifyNum)
38 : {
39 11 : CHK_PRT_RET(comm == nullptr, HCCL_ERROR("[%s] comm is null", __func__), HCCL_E_PTR);
40 9 : CHK_PRT_RET(
41 : !IsValidCommEngine(engine),
42 : HCCL_ERROR(
43 : "[%s] commEngine[%s] is invalid", __func__, GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str()),
44 : HCCL_E_PARA);
45 8 : CHK_PRT_RET(notifyNum == nullptr, HCCL_ERROR("[%s] notifyNum is null", __func__), HCCL_E_PTR);
46 :
47 6 : auto* hcclComm = static_cast<hccl::hcclComm*>(comm);
48 6 : std::string commId = hcclComm->GetIdentifier();
49 6 : HCCL_RUN_INFO(
50 : "Entry-%s:comm[%s] engine[%s]", __func__, commId.c_str(),
51 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str());
52 6 : HcclResult ret = HCCL_SUCCESS;
53 6 : if (hcclComm->IsCommunicatorV2()) {
54 3 : hccl::CollComm* collComm = hcclComm->GetCollComm();
55 3 : CHK_PTR_NULL(collComm);
56 3 : CommEngineResMgr* engineResMgr = collComm->GetCommEngineResMgr();
57 3 : CHK_PTR_NULL(engineResMgr);
58 2 : ret = engineResMgr->HcclGetNotifyNumInThread(thread, engine, notifyNum);
59 : } else {
60 3 : auto& engineResMgr = hcclComm->GetIndependentOp().GetCommEngineResMgr();
61 3 : ret = engineResMgr.HcclGetNotifyNumInThread(thread, engine, notifyNum);
62 : }
63 :
64 5 : if (ret != HCCL_SUCCESS) {
65 2 : HCCL_ERROR(
66 : "[HcclGetNotifyNumInThread] Failed to get notifyNum for engine[%s] ret[%d]",
67 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), ret);
68 2 : return ret;
69 : }
70 3 : HCCL_INFO(
71 : "[HcclGetNotifyNumInThread] threads for engine[%s], notifyNum[%u]",
72 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), *notifyNum);
73 3 : return HCCL_SUCCESS;
74 6 : }
75 :
76 19 : HcclResult HcclThreadAcquireWithConfigDfx(
77 : hccl::CollComm* collComm, const std::string& commId, CommEngine engine, u64 beginTime, uint32_t threadNum,
78 : ThreadHandle* threads, std::vector<uint32_t>& threadId)
79 : {
80 19 : CHK_PTR_NULL(threads);
81 19 : HcclCommDfx* hcclCommDfx = collComm->GetHcclCommDfx();
82 19 : CHK_PTR_NULL(hcclCommDfx);
83 19 : if (engine == CommEngine::COMM_ENGINE_AICPU) {
84 6 : Mc2CommInfo mc2CommInfo;
85 6 : mc2CommInfo.FreeStreamId = 0;
86 6 : mc2CommInfo.streamsId = threadId;
87 6 : mc2CommInfo.groupname = commId;
88 6 : mc2CommInfo.myRankId = collComm->GetMyRankId();
89 6 : mc2CommInfo.rankSize = collComm->GetRankSize();
90 6 : CHK_RET(collComm->GetParentRankId(mc2CommInfo.parentRankId));
91 6 : hcclCommDfx->ReportMc2CommInfo(mc2CommInfo);
92 6 : HCCL_INFO("[HcclThreadAcquireWithConfigDfx] ReportThreadAcquireKernel begin");
93 6 : const std::string KernelName = "RunAicpuIndOpThreadInit";
94 : // 这个地方获取不到当前是单算子还是图模式，所以全部都不保存
95 6 : CHK_RET(hcclCommDfx->ReportKernel(beginTime, commId, KernelName, SalGetTid(), false));
96 6 : HCCL_INFO("[HcclThreadAcquireWithConfigDfx] ReportThreadAcquireKernel success");
97 6 : } else {
98 13 : auto hcclCommDfxCallBack = collComm->GetDfxCallback();
99 36 : for (u32 num = 0; num < threadNum; ++num) {
100 24 : int ret = HcommThreadRegisterDfx(threads[num], hcclCommDfxCallBack);
101 24 : if (ret != HCCL_SUCCESS) {
102 1 : HCCL_ERROR(
103 : "[HcclThreadAcquireWithConfigDfx] ReportThreadAcquireKernel HcommThreadRegisterDfx failed"
104 : " ret:[%d], num:[%u]",
105 : ret, num);
106 1 : return HCCL_E_INTERNAL;
107 : }
108 : }
109 13 : }
110 18 : return HCCL_SUCCESS;
111 : }
112 :
113 : HcclResult
114 18 : ValidateThreadAcquireParams(CommEngine engine, ThreadType type, const ThreadConfig* config, uint32_t threadNum)
115 : {
116 18 : CHK_PRT_RET(
117 : type == THREAD_TYPE_INVALID,
118 : HCCL_ERROR("[%s] thread type[%d] is invalid", __func__, static_cast<int32_t>(type)), HCCL_E_PARA);
119 17 : CHK_PRT_RET(
120 : !IsValidCommEngine(engine),
121 : HCCL_ERROR("[%s] commEngine[%d] is invalid", __func__, static_cast<int32_t>(engine)), HCCL_E_PARA);
122 17 : CHK_PRT_RET(threadNum == 0, HCCL_ERROR("[%s] threadNum[%u] is invalid", __func__, threadNum), HCCL_E_PARA);
123 17 : CHK_PRT_RET(config == nullptr, HCCL_ERROR("[%s] config is null", __func__), HCCL_E_PTR);
124 32 : for (uint32_t i = 0; i < threadNum; ++i) {
125 17 : CHK_PRT_RET(
126 : config[i].header.magicWord != HCOMM_THREAD_CONFIG_MAGIC_WORD,
127 : HCCL_ERROR(
128 : "[%s] config[%u] magicWord[0x%x] mismatch, expected[0x%x], call ThreadConfigInit first", __func__, i,
129 : config[i].header.magicWord, HCOMM_THREAD_CONFIG_MAGIC_WORD),
130 : HCCL_E_PARA);
131 : }
132 15 : CHK_PRT_RET(
133 : engine == CommEngine::COMM_ENGINE_AICPU_TS || engine == CommEngine::COMM_ENGINE_CPU_TS,
134 : HCCL_ERROR(
135 : "[%s] commEngine[%d] CPU_TS/AICPU_TS not supported, use CPU/AICPU engine with THREAD_TYPE_TS instead",
136 : __func__, static_cast<int32_t>(engine)),
137 : HCCL_E_PARA);
138 13 : CHK_PRT_RET(
139 : engine == CommEngine::COMM_ENGINE_AIV || engine == CommEngine::COMM_ENGINE_CCU,
140 : HCCL_ERROR(
141 : "[%s] commEngine[%d] AIV/CCU not supported, supported engines: CPU/AICPU", __func__,
142 : static_cast<int32_t>(engine)),
143 : HCCL_E_PARA);
144 11 : return HCCL_SUCCESS;
145 : }
146 :
147 20 : HcclResult HcclThreadAcquireWithConfig(
148 : HcclComm comm, CommEngine engine, uint32_t threadNum, ThreadType type, const ThreadConfig* config,
149 : ThreadHandle* threads)
150 : {
151 20 : CHK_PRT_RET(comm == nullptr, HCCL_ERROR("[%s] comm is null", __func__), HCCL_E_PTR);
152 19 : CHK_PRT_RET(threads == nullptr, HCCL_ERROR("[%s] threads is null", __func__), HCCL_E_PTR);
153 18 : CHK_RET(ValidateThreadAcquireParams(engine, type, config, threadNum));
154 :
155 11 : u64 beginTime = Hccl::DlProfFunction::GetInstance().dlMsprofSysCycleTime();
156 11 : auto* hcclComm = static_cast<hccl::hcclComm*>(comm);
157 11 : std::string commId = hcclComm->GetIdentifier();
158 11 : HCCL_RUN_INFO(
159 : "Entry-%s:comm[%s] engine[%s] ThreadNum[%u].", __func__, commId.c_str(),
160 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), threadNum);
161 :
162 11 : HcclResult ret = HCCL_SUCCESS;
163 11 : std::vector<uint32_t> threadId;
164 11 : if (hcclComm->IsCommunicatorV2()) {
165 6 : hccl::CollComm* collComm = hcclComm->GetCollComm();
166 6 : CHK_PTR_NULL(collComm);
167 5 : CommEngineResMgr* engineResMgr = collComm->GetCommEngineResMgr();
168 5 : CHK_PTR_NULL(engineResMgr);
169 4 : ret = engineResMgr->HcclThreadAcquireV2(engine, threadNum, type, config, threads, threadId);
170 4 : if (ret != HCCL_SUCCESS) {
171 1 : HCCL_ERROR(
172 : "[%s] failed to create threads for engine[%s], threadsNum[%u], ret[%d].", __func__,
173 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), threadNum, ret);
174 1 : return ret;
175 : }
176 3 : CHK_RET(HcclThreadAcquireWithConfigDfx(collComm, commId, engine, beginTime, threadNum, threads, threadId));
177 2 : return HCCL_SUCCESS;
178 : } else {
179 5 : auto& engineResMgr = hcclComm->GetIndependentOp().GetCommEngineResMgr();
180 5 : ret = engineResMgr.HcclThreadAcquire(engine, threadNum, type, config, threads, threadId);
181 5 : if (engine == CommEngine::COMM_ENGINE_AICPU) {
182 : // 上报流
183 3 : if (threadNum != threadId.size()) {
184 1 : HCCL_ERROR("[%s] threadNum [%u] != threadId.size[%zu]", __func__, threadNum, threadId.size());
185 1 : return HCCL_E_PARA;
186 : }
187 2 : CHK_RET(HcclStreamProfilingReport(comm, threadNum, threadId.data()));
188 : }
189 : }
190 3 : if (ret != HCCL_SUCCESS) {
191 1 : HCCL_ERROR(
192 : "[%s] failed to create threads for engine[%s], threadsNum[%u], ret[%d].", __func__,
193 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), threadNum, ret);
194 1 : return ret;
195 : }
196 :
197 2 : HCCL_INFO("[%s] Allocated %u threads for engine[%d]", __func__, threadNum, engine);
198 2 : return HCCL_SUCCESS;
199 11 : }
200 :
201 32 : static CommEngine ConvertEngineToTsType(CommEngine engine)
202 : {
203 32 : if (engine == COMM_ENGINE_CPU_TS) {
204 18 : return COMM_ENGINE_CPU;
205 : }
206 14 : if (engine == COMM_ENGINE_AICPU_TS) {
207 9 : return COMM_ENGINE_AICPU;
208 : }
209 5 : return engine;
210 : }
211 :
212 33 : HcclResult HcclThreadAcquire(
213 : HcclComm comm, CommEngine engine, uint32_t threadNum, uint32_t notifyNumPerThread, ThreadHandle* threads)
214 : {
215 33 : u64 beginTime = Hccl::DlProfFunction::GetInstance().dlMsprofSysCycleTime();
216 33 : CHK_PRT_RET(comm == nullptr, HCCL_ERROR("[%s] comm is null", __func__), HCCL_E_PTR);
217 31 : CHK_PRT_RET(threads == nullptr, HCCL_ERROR("[%s] threads is null", __func__), HCCL_E_PTR);
218 29 : CHK_PRT_RET(
219 : !IsValidCommEngine(engine),
220 : HCCL_ERROR("[%s] commEngine[%d] is invalid", __func__, static_cast<int32_t>(engine)), HCCL_E_PARA);
221 28 : CHK_PRT_RET(threadNum == 0, HCCL_ERROR("[%s] threadNum[%u] is invalid", __func__, threadNum), HCCL_E_PARA);
222 :
223 28 : auto* hcclComm = static_cast<hccl::hcclComm*>(comm);
224 28 : std::string commId = hcclComm->GetIdentifier();
225 28 : HCCL_RUN_INFO(
226 : "Entry-%s:comm[%s] engine[%u] ThreadNum[%u] notifyNumPerThread[%u]", __func__, commId.c_str(), engine,
227 : threadNum, notifyNumPerThread);
228 :
229 28 : CommEngine newEngine = ConvertEngineToTsType(engine);
230 28 : ThreadType type = THREAD_TYPE_TS;
231 28 : std::unique_ptr<ThreadConfig[]> config = std::make_unique<ThreadConfig[]>(threadNum);
232 28 : CHK_PTR_NULL(config);
233 28 : CHK_PRT_RET(
234 : ThreadConfigInit(config.get(), threadNum) != 0, HCCL_ERROR("[%s] ThreadConfigInit failed", __func__),
235 : HCCL_E_INTERNAL);
236 28 : CHK_PRT_RET(
237 : notifyNumPerThread >= HCCL_THREAD_NOTIFY_MAX_NUM,
238 : HCCL_ERROR("[%s] notifyNumPerThread[%u] exceeds HCCL_THREAD_NOTIFY_MAX_NUM", __func__, notifyNumPerThread),
239 : HCCL_E_PARA);
240 471 : for (u32 i = 0; i < threadNum; i++) {
241 445 : config[i].notifyNumPerThread = static_cast<uint16_t>(notifyNumPerThread);
242 : }
243 :
244 26 : HcclResult ret = HCCL_SUCCESS;
245 26 : std::vector<uint32_t> threadId;
246 26 : if (hcclComm->IsCommunicatorV2()) {
247 20 : hccl::CollComm* collComm = hcclComm->GetCollComm();
248 20 : CHK_PTR_NULL(collComm);
249 19 : CommEngineResMgr* engineResMgr = collComm->GetCommEngineResMgr();
250 19 : CHK_PTR_NULL(engineResMgr);
251 18 : ret = engineResMgr->HcclThreadAcquireV2(newEngine, threadNum, type, config.get(), threads, threadId);
252 18 : if (ret != HCCL_SUCCESS) {
253 2 : HCCL_ERROR(
254 : "[%s] failed to create threads for engine[%d], threadsNum[%u], ret[%d]", __func__, newEngine, threadNum,
255 : ret);
256 2 : return ret;
257 : }
258 16 : CHK_RET(HcclThreadAcquireWithConfigDfx(collComm, commId, newEngine, beginTime, threadNum, threads, threadId));
259 : } else {
260 6 : auto& engineResMgr = hcclComm->GetIndependentOp().GetCommEngineResMgr();
261 6 : ret = engineResMgr.HcclThreadAcquire(newEngine, threadNum, type, config.get(), threads, threadId);
262 6 : if (newEngine == CommEngine::COMM_ENGINE_AICPU) {
263 0 : if (threadNum != threadId.size()) {
264 0 : HCCL_ERROR("[%s] threadNum [%u] != threadId.size[%zu]", __func__, threadNum, threadId.size());
265 0 : return HCCL_E_PARA;
266 : }
267 0 : CHK_RET(HcclStreamProfilingReport(comm, threadNum, threadId.data()));
268 : }
269 : }
270 22 : if (ret != HCCL_SUCCESS) {
271 2 : HCCL_ERROR(
272 : "[%s] Failed to create threads for engine[%s], threadNum[%u], ret[%d]", __func__,
273 : GetEnumToString(GetCommEngineStatusStrMap(), newEngine).c_str(), threadNum, ret);
274 2 : return ret;
275 : }
276 :
277 20 : HCCL_INFO(
278 : "[%s] Allocated %u threads for engine[%s], notifyPerThread[%u]", __func__, threadNum,
279 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), notifyNumPerThread);
280 20 : return HCCL_SUCCESS;
281 28 : }
282 :
283 1 : HcclResult HcclThreadAcquireWithStreamDfx(
284 : hccl::CollComm* collComm, const std::string& commId, CommEngine engine, ThreadHandle thread)
285 : {
286 1 : auto hcclCommDfxCallback = collComm->GetDfxCallback();
287 1 : int ret = HcommThreadRegisterDfx(thread, hcclCommDfxCallback);
288 1 : if (ret != 0) {
289 0 : HCCL_ERROR("[HcclThreadAcquire] HcclThreadAcquire HcommThreadRegisterDfx failed, ret:[%d]", ret);
290 0 : return HCCL_E_INTERNAL;
291 : }
292 1 : if (engine == CommEngine::COMM_ENGINE_AICPU) {
293 0 : Thread* threadPtr = reinterpret_cast<Thread*>(thread);
294 0 : CHK_PTR_NULL(threadPtr);
295 0 : Stream* threadStream = threadPtr->GetStream();
296 0 : CHK_PTR_NULL(threadStream);
297 0 : Mc2CommInfo mc2CommInfo;
298 0 : mc2CommInfo.FreeStreamId = 0;
299 0 : mc2CommInfo.streamsId.push_back(static_cast<u32>(threadStream->sqId()));
300 0 : mc2CommInfo.groupname = commId;
301 0 : mc2CommInfo.myRankId = collComm->GetMyRankId();
302 0 : mc2CommInfo.rankSize = collComm->GetRankSize();
303 0 : CHK_RET(collComm->GetParentRankId(mc2CommInfo.parentRankId));
304 0 : HcclCommDfx* hcclCommDfx = collComm->GetHcclCommDfx();
305 0 : CHK_PTR_NULL(hcclCommDfx);
306 0 : hcclCommDfx->ReportMc2CommInfo(mc2CommInfo);
307 0 : }
308 1 : return HCCL_SUCCESS;
309 1 : }
310 :
311 7 : HcclResult HcclThreadAcquireWithStream(
312 : HcclComm comm, CommEngine engine, aclrtStream stream, uint32_t notifyNum, ThreadHandle* thread)
313 : {
314 7 : CHK_PTR_NULL(comm);
315 6 : CHK_PTR_NULL(stream);
316 5 : CHK_PTR_NULL(thread);
317 4 : CHK_PRT_RET(
318 : !IsValidCommEngine(engine),
319 : HCCL_ERROR("[%s] commEngine[%d] is invalid", __func__, static_cast<int32_t>(engine)), HCCL_E_PARA);
320 :
321 4 : CommEngine newEngine = ConvertEngineToTsType(engine);
322 :
323 4 : auto* hcclComm = static_cast<hccl::hcclComm*>(comm);
324 4 : std::string commId = hcclComm->GetIdentifier();
325 4 : HCCL_INFO(
326 : "Entry-%s:comm[%s] engine[%s] notifyNum[%u] stream[%p]", __func__, commId.c_str(),
327 : GetEnumToString(GetCommEngineStatusStrMap(), newEngine).c_str(), notifyNum, stream);
328 4 : HcclResult ret = HCCL_SUCCESS;
329 4 : if (hcclComm->IsCommunicatorV2()) {
330 2 : hccl::CollComm* collComm = hcclComm->GetCollComm();
331 2 : CHK_PTR_NULL(collComm);
332 2 : CommEngineResMgr* engineResMgr = collComm->GetCommEngineResMgr();
333 2 : CHK_PTR_NULL(engineResMgr);
334 1 : ret = engineResMgr->HcclThreadAcquireWithStream(newEngine, stream, notifyNum, thread);
335 1 : CHK_RET(HcclThreadAcquireWithStreamDfx(collComm, commId, newEngine, *thread));
336 : } else {
337 2 : auto& engineResMgr = hcclComm->GetIndependentOp().GetCommEngineResMgr();
338 2 : ret = engineResMgr.HcclThreadAcquireWithStream(newEngine, stream, notifyNum, thread);
339 : }
340 :
341 3 : if (ret != HCCL_SUCCESS) {
342 1 : HCCL_ERROR(
343 : "[HcclThreadAcquireWithStream] Failed to create thread for engine[%s], ret[%d]",
344 : GetEnumToString(GetCommEngineStatusStrMap(), newEngine).c_str(), ret);
345 1 : return ret;
346 : }
347 :
348 2 : HCCL_INFO(
349 : "[HcclThreadAcquireWithStream] Allocated thread for engine[%s], stream[%p], notifyNum[%u]",
350 : GetEnumToString(GetCommEngineStatusStrMap(), newEngine).c_str(), stream, notifyNum);
351 2 : return HCCL_SUCCESS;
352 4 : }
353 :
354 13 : HcclResult HcclDedicatedThreadAcquire(
355 : HcclComm comm, HcclDedicatedThreadType useType, uint32_t notifyNumPerThread, ThreadHandle* thread)
356 : {
357 : EXCEPTION_HANDLE_BEGIN
358 15 : CHK_PRT_RET(comm == nullptr, HCCL_ERROR("[%s] comm is null", __func__), HCCL_E_PTR);
359 12 : CHK_PRT_RET(thread == nullptr, HCCL_ERROR("[%s] thread is null", __func__), HCCL_E_PTR);
360 11 : CHK_PRT_RET(
361 : useType == HCCL_DED_THREAD_TYPE_INVALID, HCCL_ERROR("[%s] dedThreadType is invalid", __func__), HCCL_E_PARA);
362 :
363 10 : auto* hcclComm = static_cast<hccl::hcclComm*>(comm);
364 10 : const std::string& commId = hcclComm->GetIdentifier();
365 10 : HCCL_INFO(
366 : "Entry-%s:comm[%s] dedThreadType[%u] notifyNumPerThread[%u]", __func__, commId.c_str(), useType,
367 : notifyNumPerThread);
368 10 : hccl::CollComm* collComm = hcclComm->GetCollComm();
369 10 : CHK_PTR_NULL(collComm);
370 : /* 保序场景：委托给 OrderLaunchThreadMgr（进程粒度） */
371 10 : if (ORDER_LAUNCH_TYPES.find(useType) != ORDER_LAUNCH_TYPES.end()) {
372 0 : s32 deviceLogicId = Hccl::HrtGetDevice();
373 0 : auto& resMgr = hccl::CollCommMgr::GetInstance().GetOrderLaunchThreadMgr(deviceLogicId);
374 0 : ThreadHandle th = 0;
375 0 : HcclResult ret = resMgr.OrderLaunchThreadAcquire(useType, collComm, commId, notifyNumPerThread, th);
376 0 : CHK_PRT_RET(
377 : ret != HCCL_SUCCESS,
378 : HCCL_ERROR(
379 : "[%s] OrderLaunchThreadAcquire fail, ret[%d], useType[%d]", __func__, ret, static_cast<s32>(useType)),
380 : ret);
381 0 : *thread = th;
382 0 : return HCCL_SUCCESS;
383 : }
384 :
385 10 : CommEngineResMgr* engineResMgr = collComm->GetCommEngineResMgr();
386 10 : CHK_PTR_NULL(engineResMgr);
387 10 : CHK_RET(engineResMgr->HcclDedicatedThreadAcquire(useType, notifyNumPerThread, thread));
388 10 : HCCL_INFO(
389 : "[%s] success, dedThreadType[%u], thread[0x%llx], notifyNumPerThread[%u]", __func__, useType, *thread,
390 : notifyNumPerThread);
391 10 : EXCEPTION_HANDLE_END
392 :
393 10 : return HCCL_SUCCESS;
394 : }
395 :
396 0 : HcclResult HcclAllocNotify(
397 : HcclComm comm, CommEngine commEngine, ::NotifyType notifyType, uint32_t notifyNum, NotifyHandle** notifyHandleList)
398 : {
399 0 : CHK_PRT_RET(comm == nullptr, HCCL_ERROR("[%s] comm is null", __func__), HCCL_E_PARA);
400 0 : CHK_PRT_RET(
401 : !IsValidCommEngine(commEngine),
402 : HCCL_ERROR(
403 : "[%s] commEngine[%s] is invalid", __func__,
404 : GetEnumToString(GetCommEngineStatusStrMap(), commEngine).c_str()),
405 : HCCL_E_PARA);
406 0 : CHK_PRT_RET(
407 : !IsValidNotify(notifyType), HCCL_ERROR("[%s] notifyType[%u] is invalid", __func__, notifyType), HCCL_E_PARA);
408 0 : CHK_PRT_RET(
409 : notifyNum > NOTIFY_MAX_NUM || notifyNum == 0, HCCL_ERROR("[%s] notifyNum[%u] is invalid", __func__, notifyNum),
410 : HCCL_E_PARA);
411 0 : CHK_PRT_RET(notifyHandleList == nullptr, HCCL_ERROR("[%s] notifyHandleList is null", __func__), HCCL_E_PARA);
412 0 : CHK_PRT_RET(*notifyHandleList != nullptr, HCCL_ERROR("[%s] notifyHandleList is not null", __func__), HCCL_E_PARA);
413 :
414 0 : if (commEngine == CommEngine::COMM_ENGINE_CPU || commEngine == CommEngine::COMM_ENGINE_CPU_TS
415 0 : || commEngine == CommEngine::COMM_ENGINE_CCU) {
416 0 : if (notifyType != ::NOTIFY_TYPE_RTS_NOTIFY) {
417 0 : HCCL_ERROR(
418 : "[%s] commEngine[%s] and notifyType[%u] are mismatch", __func__,
419 : GetEnumToString(GetCommEngineStatusStrMap(), commEngine).c_str(), notifyType);
420 0 : return HCCL_E_PARA;
421 : }
422 : } else {
423 0 : if (notifyType != ::NOTIFY_TYPE_DEVICE_MEM) {
424 0 : HCCL_ERROR(
425 : "[%s] commEngine[%s] and notifyType[%u] are mismatch", __func__,
426 : GetEnumToString(GetCommEngineStatusStrMap(), commEngine).c_str(), notifyType);
427 0 : return HCCL_E_PARA;
428 : }
429 : }
430 :
431 0 : auto* hcclComm = static_cast<hccl::hcclComm*>(comm);
432 0 : std::string commId = hcclComm->GetIdentifier();
433 0 : HCCL_RUN_INFO(
434 : "Entry-%s:comm[%s] commEngine[%s] notifyType[%u] notifyNum[%u]", __func__, commId.c_str(),
435 : GetEnumToString(GetCommEngineStatusStrMap(), commEngine).c_str(), notifyType, notifyNum);
436 0 : HcclResult ret = HCCL_SUCCESS;
437 0 : if (hcclComm->IsCommunicatorV2()) {
438 0 : hccl::CollComm* collComm = hcclComm->GetCollComm();
439 0 : CHK_PTR_NULL(collComm);
440 0 : CommEngineResMgr* engineResMgr = collComm->GetCommEngineResMgr();
441 0 : CHK_PTR_NULL(engineResMgr);
442 0 : ret = engineResMgr->HcclAllocNotify(commEngine, notifyType, notifyNum, notifyHandleList);
443 : } else {
444 0 : auto& engineResMgr = hcclComm->GetIndependentOp().GetCommEngineResMgr();
445 0 : ret = engineResMgr.HcclAllocNotify(commEngine, notifyType, notifyNum, notifyHandleList);
446 : }
447 :
448 0 : if (ret != HCCL_SUCCESS) {
449 0 : HCCL_ERROR(
450 : "[%s] Failed to create notify for commEngine[%s]", __func__,
451 : GetEnumToString(GetCommEngineStatusStrMap(), commEngine).c_str());
452 0 : return ret;
453 : }
454 :
455 0 : HCCL_RUN_INFO(
456 : "[%s] Allocated notify for commEngine[%s], notifyType[%u], notifyNum[%u]", __func__,
457 : GetEnumToString(GetCommEngineStatusStrMap(), commEngine).c_str(), notifyType, notifyNum);
458 0 : return HCCL_SUCCESS;
459 0 : }
460 :
461 0 : HcclResult HcommFreeNotify(HcclComm comm, uint32_t notifyNum, NotifyHandle* notifyHandleList)
462 : {
463 0 : CHK_PRT_RET(comm == nullptr, HCCL_ERROR("[%s] comm is null", __func__), HCCL_E_PARA);
464 0 : CHK_PRT_RET(notifyHandleList == nullptr, HCCL_ERROR("[%s] notifyHandleList is null", __func__), HCCL_E_PARA);
465 0 : CHK_PRT_RET(
466 : notifyNum > NOTIFY_MAX_NUM || notifyNum == 0, HCCL_ERROR("[%s] notifyNum[%u] is invalid", __func__, notifyNum),
467 : HCCL_E_PARA);
468 0 : auto* hcclComm = static_cast<hccl::hcclComm*>(comm);
469 0 : std::string commId = hcclComm->GetIdentifier();
470 0 : HCCL_RUN_INFO("Entry-%s:comm[%s] notifyNum[%u]", __func__, commId.c_str(), notifyNum);
471 0 : HcclResult ret = HCCL_SUCCESS;
472 0 : if (hcclComm->IsCommunicatorV2()) {
473 0 : hccl::CollComm* collComm = hcclComm->GetCollComm();
474 0 : CHK_PTR_NULL(collComm);
475 0 : CommEngineResMgr* engineResMgr = collComm->GetCommEngineResMgr();
476 0 : CHK_PTR_NULL(engineResMgr);
477 0 : ret = engineResMgr->HcommFreeNotify(notifyNum, notifyHandleList);
478 : } else {
479 0 : auto& engineResMgr = hcclComm->GetIndependentOp().GetCommEngineResMgr();
480 0 : ret = engineResMgr.HcommFreeNotify(notifyNum, notifyHandleList);
481 : }
482 0 : if (ret != HCCL_SUCCESS) {
483 0 : HCCL_ERROR("[%s] Failed to free notify", __func__);
484 0 : return ret;
485 : }
486 :
487 0 : HCCL_RUN_INFO("[%s] Free notify for notifyNum[%u]", __func__, notifyNum);
488 0 : return HCCL_SUCCESS;
489 0 : }
490 :
491 : #ifdef __cplusplus
492 : extern "C" {
493 : #endif
494 11 : HcclResult HcclThreadExportToCommEngine(
495 : HcclComm comm, uint32_t threadNum, const ThreadHandle* threads, CommEngine dstCommEngine,
496 : ThreadHandle* exportedThreads)
497 : {
498 11 : CHK_PTR_NULL(comm);
499 10 : CHK_PTR_NULL(threads);
500 9 : CHK_PTR_NULL(exportedThreads);
501 8 : CHK_PRT_RET(
502 : !IsValidCommEngine(dstCommEngine),
503 : HCCL_ERROR(
504 : "[%s] commEngine[%s] is invalid", __func__,
505 : GetEnumToString(GetCommEngineStatusStrMap(), dstCommEngine).c_str()),
506 : HCCL_E_PARA);
507 7 : if (threadNum == 0 || threadNum > MAX_EXPORT_THREAD_NUM) {
508 2 : HCCL_ERROR("[%s] threadNum[%u] is 0 or greater than %u", __func__, threadNum, MAX_EXPORT_THREAD_NUM);
509 2 : return HCCL_E_PARA;
510 : }
511 :
512 5 : hccl::hcclComm* hcclComm = static_cast<hccl::hcclComm*>(comm);
513 5 : std::string commId = hcclComm->GetIdentifier();
514 5 : HCCL_INFO(
515 : "Entry-[%s]:comm[%s], threadNum[%u], commEngine[%s], threadsPtr[%p], exportedThreadsPtr[%p]", __func__,
516 : commId.c_str(), threadNum, GetEnumToString(GetCommEngineStatusStrMap(), dstCommEngine).c_str(), threads,
517 : exportedThreads);
518 : HcclResult ret;
519 5 : if (hcclComm->IsCommunicatorV2()) {
520 1 : hccl::CollComm* collComm = hcclComm->GetCollComm();
521 1 : CHK_PTR_NULL(collComm);
522 1 : CommEngineResMgr* engineResMgr = collComm->GetCommEngineResMgr();
523 1 : CHK_PTR_NULL(engineResMgr);
524 1 : ret = engineResMgr->HcclThreadExportToCommEngine(threadNum, threads, dstCommEngine, exportedThreads);
525 : } else {
526 4 : auto& engineResMgr = hcclComm->GetIndependentOp().GetCommEngineResMgr();
527 4 : ret = engineResMgr.HcclThreadExportToCommEngine(threadNum, threads, dstCommEngine, exportedThreads);
528 : }
529 :
530 5 : CHK_PRT_RET(
531 : ret != HCCL_SUCCESS,
532 : HCCL_ERROR(
533 : "[%s] Thread export failed. Export threadNum[%u], commEngine[%s], threadsPtr[%p], exportedThreadsPtr[%p]",
534 : __func__, threadNum, GetEnumToString(GetCommEngineStatusStrMap(), dstCommEngine).c_str(), threads,
535 : exportedThreads),
536 : ret);
537 3 : HCCL_INFO("[%s]:comm[%s] export success.", __func__, commId.c_str());
538 3 : return HCCL_SUCCESS;
539 5 : }
540 : #ifdef __cplusplus
541 : }
542 : #endif
543 :
544 : HcclResult
545 6 : HcclThreadResGetInfo(HcclComm comm, ThreadHandle thread, ThreadResType resType, uint32_t infoLen, void** info)
546 : {
547 6 : CHK_PTR_NULL(comm);
548 5 : CHK_PTR_NULL(info);
549 4 : hccl::hcclComm* hcclComm = static_cast<hccl::hcclComm*>(comm);
550 4 : std::string commId = hcclComm->GetIdentifier();
551 4 : HCCL_INFO(
552 : "Entry-[%s]:comm[%s], thread[0x%llx], resType[%d], infoLen[%u], info[%p]", __func__, commId.c_str(),
553 : static_cast<unsigned long long>(thread), static_cast<int32_t>(resType), infoLen, info);
554 4 : HcclResult ret = HCCL_SUCCESS;
555 4 : if (hcclComm->IsCommunicatorV2()) {
556 4 : hccl::CollComm* collComm = hcclComm->GetCollComm();
557 4 : CHK_PTR_NULL(collComm);
558 4 : CommEngineResMgr* engineResMgr = collComm->GetCommEngineResMgr();
559 4 : CHK_PTR_NULL(engineResMgr);
560 4 : ret = engineResMgr->HcclThreadResGetInfo(thread, resType, infoLen, info);
561 : } else {
562 : DevType devType;
563 0 : CHK_RET(hrtGetDeviceType(devType));
564 0 : if (devType != DevType::DEV_TYPE_910B) { // 910B HOST网卡需要走此流程，不打印错误日志
565 0 : HCCL_ERROR("[%s] communicatorType is not supported.", __func__);
566 : }
567 0 : return HCCL_E_NOT_SUPPORT;
568 : }
569 4 : CHK_PRT_RET(
570 : ret != HCCL_SUCCESS,
571 : HCCL_ERROR(
572 : "[%s] thread resource get info failed. thread[0x%llx], resType[%d], infoLen[%u], info[%p]", __func__,
573 : static_cast<unsigned long long>(thread), static_cast<int32_t>(resType), infoLen, info),
574 : ret);
575 1 : HCCL_INFO("[%s]:comm[%s] get thread resource success.", __func__, commId.c_str());
576 1 : return HCCL_SUCCESS;
577 4 : }
|
