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_PRT_RET(
336 : ret != HCCL_SUCCESS, HCCL_ERROR("[%s] HcclThreadAcquireWithStream failed, ret[%d]", __func__, ret), ret);
337 1 : CHK_RET(HcclThreadAcquireWithStreamDfx(collComm, commId, newEngine, *thread));
338 : } else {
339 2 : auto& engineResMgr = hcclComm->GetIndependentOp().GetCommEngineResMgr();
340 2 : ret = engineResMgr.HcclThreadAcquireWithStream(newEngine, stream, notifyNum, thread);
341 2 : CHK_PRT_RET(
342 : ret != HCCL_SUCCESS, HCCL_ERROR("[%s] HcclThreadAcquireWithStream failed, ret[%d]", __func__, ret), ret);
343 : }
344 :
345 2 : HCCL_INFO(
346 : "[HcclThreadAcquireWithStream] Allocated thread for engine[%s], stream[%p], notifyNum[%u]",
347 : GetEnumToString(GetCommEngineStatusStrMap(), newEngine).c_str(), stream, notifyNum);
348 2 : return HCCL_SUCCESS;
349 4 : }
350 :
351 13 : HcclResult HcclDedicatedThreadAcquire(
352 : HcclComm comm, HcclDedicatedThreadType useType, uint32_t notifyNumPerThread, ThreadHandle* thread)
353 : {
354 : EXCEPTION_HANDLE_BEGIN
355 15 : CHK_PRT_RET(comm == nullptr, HCCL_ERROR("[%s] comm is null", __func__), HCCL_E_PTR);
356 12 : CHK_PRT_RET(thread == nullptr, HCCL_ERROR("[%s] thread is null", __func__), HCCL_E_PTR);
357 11 : CHK_PRT_RET(
358 : useType == HCCL_DED_THREAD_TYPE_INVALID, HCCL_ERROR("[%s] dedThreadType is invalid", __func__), HCCL_E_PARA);
359 :
360 10 : auto* hcclComm = static_cast<hccl::hcclComm*>(comm);
361 10 : const std::string& commId = hcclComm->GetIdentifier();
362 10 : HCCL_INFO(
363 : "Entry-%s:comm[%s] dedThreadType[%u] notifyNumPerThread[%u]", __func__, commId.c_str(), useType,
364 : notifyNumPerThread);
365 10 : hccl::CollComm* collComm = hcclComm->GetCollComm();
366 10 : CHK_PTR_NULL(collComm);
367 : /* 保序场景：委托给 OrderLaunchThreadMgr（进程粒度） */
368 10 : if (ORDER_LAUNCH_TYPES.find(useType) != ORDER_LAUNCH_TYPES.end()) {
369 0 : s32 deviceLogicId = Hccl::HrtGetDevice();
370 0 : auto& resMgr = hccl::CollCommMgr::GetInstance().GetOrderLaunchThreadMgr(deviceLogicId);
371 0 : ThreadHandle th = 0;
372 0 : HcclResult ret = resMgr.OrderLaunchThreadAcquire(useType, collComm, commId, notifyNumPerThread, th);
373 0 : CHK_PRT_RET(
374 : ret != HCCL_SUCCESS,
375 : HCCL_ERROR(
376 : "[%s] OrderLaunchThreadAcquire fail, ret[%d], useType[%d]", __func__, ret, static_cast<s32>(useType)),
377 : ret);
378 0 : *thread = th;
379 0 : return HCCL_SUCCESS;
380 : }
381 :
382 10 : CommEngineResMgr* engineResMgr = collComm->GetCommEngineResMgr();
383 10 : CHK_PTR_NULL(engineResMgr);
384 10 : CHK_RET(engineResMgr->HcclDedicatedThreadAcquire(useType, notifyNumPerThread, thread));
385 10 : HCCL_INFO(
386 : "[%s] success, dedThreadType[%u], thread[0x%llx], notifyNumPerThread[%u]", __func__, useType, *thread,
387 : notifyNumPerThread);
388 10 : EXCEPTION_HANDLE_END
389 :
390 10 : return HCCL_SUCCESS;
391 : }
392 :
393 0 : HcclResult HcclAllocNotify(
394 : HcclComm comm, CommEngine commEngine, ::NotifyType notifyType, uint32_t notifyNum, NotifyHandle** notifyHandleList)
395 : {
396 0 : CHK_PRT_RET(comm == nullptr, HCCL_ERROR("[%s] comm is null", __func__), HCCL_E_PARA);
397 0 : CHK_PRT_RET(
398 : !IsValidCommEngine(commEngine),
399 : HCCL_ERROR(
400 : "[%s] commEngine[%s] is invalid", __func__,
401 : GetEnumToString(GetCommEngineStatusStrMap(), commEngine).c_str()),
402 : HCCL_E_PARA);
403 0 : CHK_PRT_RET(
404 : !IsValidNotify(notifyType), HCCL_ERROR("[%s] notifyType[%u] is invalid", __func__, notifyType), HCCL_E_PARA);
405 0 : CHK_PRT_RET(
406 : notifyNum > NOTIFY_MAX_NUM || notifyNum == 0, HCCL_ERROR("[%s] notifyNum[%u] is invalid", __func__, notifyNum),
407 : HCCL_E_PARA);
408 0 : CHK_PRT_RET(notifyHandleList == nullptr, HCCL_ERROR("[%s] notifyHandleList is null", __func__), HCCL_E_PARA);
409 0 : CHK_PRT_RET(*notifyHandleList != nullptr, HCCL_ERROR("[%s] notifyHandleList is not null", __func__), HCCL_E_PARA);
410 :
411 0 : if (commEngine == CommEngine::COMM_ENGINE_CPU || commEngine == CommEngine::COMM_ENGINE_CPU_TS
412 0 : || commEngine == CommEngine::COMM_ENGINE_CCU) {
413 0 : if (notifyType != ::NOTIFY_TYPE_RTS_NOTIFY) {
414 0 : HCCL_ERROR(
415 : "[%s] commEngine[%s] and notifyType[%u] are mismatch", __func__,
416 : GetEnumToString(GetCommEngineStatusStrMap(), commEngine).c_str(), notifyType);
417 0 : return HCCL_E_PARA;
418 : }
419 : } else {
420 0 : if (notifyType != ::NOTIFY_TYPE_DEVICE_MEM) {
421 0 : HCCL_ERROR(
422 : "[%s] commEngine[%s] and notifyType[%u] are mismatch", __func__,
423 : GetEnumToString(GetCommEngineStatusStrMap(), commEngine).c_str(), notifyType);
424 0 : return HCCL_E_PARA;
425 : }
426 : }
427 :
428 0 : auto* hcclComm = static_cast<hccl::hcclComm*>(comm);
429 0 : std::string commId = hcclComm->GetIdentifier();
430 0 : HCCL_RUN_INFO(
431 : "Entry-%s:comm[%s] commEngine[%s] notifyType[%u] notifyNum[%u]", __func__, commId.c_str(),
432 : GetEnumToString(GetCommEngineStatusStrMap(), commEngine).c_str(), notifyType, notifyNum);
433 0 : HcclResult ret = HCCL_SUCCESS;
434 0 : if (hcclComm->IsCommunicatorV2()) {
435 0 : hccl::CollComm* collComm = hcclComm->GetCollComm();
436 0 : CHK_PTR_NULL(collComm);
437 0 : CommEngineResMgr* engineResMgr = collComm->GetCommEngineResMgr();
438 0 : CHK_PTR_NULL(engineResMgr);
439 0 : ret = engineResMgr->HcclAllocNotify(commEngine, notifyType, notifyNum, notifyHandleList);
440 : } else {
441 0 : auto& engineResMgr = hcclComm->GetIndependentOp().GetCommEngineResMgr();
442 0 : ret = engineResMgr.HcclAllocNotify(commEngine, notifyType, notifyNum, notifyHandleList);
443 : }
444 :
445 0 : if (ret != HCCL_SUCCESS) {
446 0 : HCCL_ERROR(
447 : "[%s] Failed to create notify for commEngine[%s]", __func__,
448 : GetEnumToString(GetCommEngineStatusStrMap(), commEngine).c_str());
449 0 : return ret;
450 : }
451 :
452 0 : HCCL_RUN_INFO(
453 : "[%s] Allocated notify for commEngine[%s], notifyType[%u], notifyNum[%u]", __func__,
454 : GetEnumToString(GetCommEngineStatusStrMap(), commEngine).c_str(), notifyType, notifyNum);
455 0 : return HCCL_SUCCESS;
456 0 : }
457 :
458 0 : HcclResult HcommFreeNotify(HcclComm comm, uint32_t notifyNum, NotifyHandle* notifyHandleList)
459 : {
460 0 : CHK_PRT_RET(comm == nullptr, HCCL_ERROR("[%s] comm is null", __func__), HCCL_E_PARA);
461 0 : CHK_PRT_RET(notifyHandleList == nullptr, HCCL_ERROR("[%s] notifyHandleList is null", __func__), HCCL_E_PARA);
462 0 : CHK_PRT_RET(
463 : notifyNum > NOTIFY_MAX_NUM || notifyNum == 0, HCCL_ERROR("[%s] notifyNum[%u] is invalid", __func__, notifyNum),
464 : HCCL_E_PARA);
465 0 : auto* hcclComm = static_cast<hccl::hcclComm*>(comm);
466 0 : std::string commId = hcclComm->GetIdentifier();
467 0 : HCCL_RUN_INFO("Entry-%s:comm[%s] notifyNum[%u]", __func__, commId.c_str(), notifyNum);
468 0 : HcclResult ret = HCCL_SUCCESS;
469 0 : if (hcclComm->IsCommunicatorV2()) {
470 0 : hccl::CollComm* collComm = hcclComm->GetCollComm();
471 0 : CHK_PTR_NULL(collComm);
472 0 : CommEngineResMgr* engineResMgr = collComm->GetCommEngineResMgr();
473 0 : CHK_PTR_NULL(engineResMgr);
474 0 : ret = engineResMgr->HcommFreeNotify(notifyNum, notifyHandleList);
475 : } else {
476 0 : auto& engineResMgr = hcclComm->GetIndependentOp().GetCommEngineResMgr();
477 0 : ret = engineResMgr.HcommFreeNotify(notifyNum, notifyHandleList);
478 : }
479 0 : if (ret != HCCL_SUCCESS) {
480 0 : HCCL_ERROR("[%s] Failed to free notify", __func__);
481 0 : return ret;
482 : }
483 :
484 0 : HCCL_RUN_INFO("[%s] Free notify for notifyNum[%u]", __func__, notifyNum);
485 0 : return HCCL_SUCCESS;
486 0 : }
487 :
488 : #ifdef __cplusplus
489 : extern "C" {
490 : #endif
491 11 : HcclResult HcclThreadExportToCommEngine(
492 : HcclComm comm, uint32_t threadNum, const ThreadHandle* threads, CommEngine dstCommEngine,
493 : ThreadHandle* exportedThreads)
494 : {
495 11 : CHK_PTR_NULL(comm);
496 10 : CHK_PTR_NULL(threads);
497 9 : CHK_PTR_NULL(exportedThreads);
498 8 : CHK_PRT_RET(
499 : !IsValidCommEngine(dstCommEngine),
500 : HCCL_ERROR(
501 : "[%s] commEngine[%s] is invalid", __func__,
502 : GetEnumToString(GetCommEngineStatusStrMap(), dstCommEngine).c_str()),
503 : HCCL_E_PARA);
504 7 : if (threadNum == 0 || threadNum > MAX_EXPORT_THREAD_NUM) {
505 2 : HCCL_ERROR("[%s] threadNum[%u] is 0 or greater than %u", __func__, threadNum, MAX_EXPORT_THREAD_NUM);
506 2 : return HCCL_E_PARA;
507 : }
508 :
509 5 : hccl::hcclComm* hcclComm = static_cast<hccl::hcclComm*>(comm);
510 5 : std::string commId = hcclComm->GetIdentifier();
511 5 : HCCL_INFO(
512 : "Entry-[%s]:comm[%s], threadNum[%u], commEngine[%s], threadsPtr[%p], exportedThreadsPtr[%p]", __func__,
513 : commId.c_str(), threadNum, GetEnumToString(GetCommEngineStatusStrMap(), dstCommEngine).c_str(), threads,
514 : exportedThreads);
515 : HcclResult ret;
516 5 : if (hcclComm->IsCommunicatorV2()) {
517 1 : hccl::CollComm* collComm = hcclComm->GetCollComm();
518 1 : CHK_PTR_NULL(collComm);
519 1 : CommEngineResMgr* engineResMgr = collComm->GetCommEngineResMgr();
520 1 : CHK_PTR_NULL(engineResMgr);
521 1 : ret = engineResMgr->HcclThreadExportToCommEngine(threadNum, threads, dstCommEngine, exportedThreads);
522 : } else {
523 4 : auto& engineResMgr = hcclComm->GetIndependentOp().GetCommEngineResMgr();
524 4 : ret = engineResMgr.HcclThreadExportToCommEngine(threadNum, threads, dstCommEngine, exportedThreads);
525 : }
526 :
527 5 : CHK_PRT_RET(
528 : ret != HCCL_SUCCESS,
529 : HCCL_ERROR(
530 : "[%s] Thread export failed. Export threadNum[%u], commEngine[%s], threadsPtr[%p], exportedThreadsPtr[%p]",
531 : __func__, threadNum, GetEnumToString(GetCommEngineStatusStrMap(), dstCommEngine).c_str(), threads,
532 : exportedThreads),
533 : ret);
534 3 : HCCL_INFO("[%s]:comm[%s] export success.", __func__, commId.c_str());
535 3 : return HCCL_SUCCESS;
536 5 : }
537 : #ifdef __cplusplus
538 : }
539 : #endif
540 :
541 : HcclResult
542 6 : HcclThreadResGetInfo(HcclComm comm, ThreadHandle thread, ThreadResType resType, uint32_t infoLen, void** info)
543 : {
544 6 : CHK_PTR_NULL(comm);
545 5 : CHK_PTR_NULL(info);
546 4 : hccl::hcclComm* hcclComm = static_cast<hccl::hcclComm*>(comm);
547 4 : std::string commId = hcclComm->GetIdentifier();
548 4 : HCCL_INFO(
549 : "Entry-[%s]:comm[%s], thread[0x%llx], resType[%d], infoLen[%u], info[%p]", __func__, commId.c_str(),
550 : static_cast<unsigned long long>(thread), static_cast<int32_t>(resType), infoLen, info);
551 4 : HcclResult ret = HCCL_SUCCESS;
552 4 : if (hcclComm->IsCommunicatorV2()) {
553 4 : hccl::CollComm* collComm = hcclComm->GetCollComm();
554 4 : CHK_PTR_NULL(collComm);
555 4 : CommEngineResMgr* engineResMgr = collComm->GetCommEngineResMgr();
556 4 : CHK_PTR_NULL(engineResMgr);
557 4 : ret = engineResMgr->HcclThreadResGetInfo(thread, resType, infoLen, info);
558 : } else {
559 : DevType devType;
560 0 : CHK_RET(hrtGetDeviceType(devType));
561 0 : if (devType != DevType::DEV_TYPE_910B) { // 910B HOST网卡需要走此流程，不打印错误日志
562 0 : HCCL_ERROR("[%s] communicatorType is not supported.", __func__);
563 : }
564 0 : return HCCL_E_NOT_SUPPORT;
565 : }
566 4 : CHK_PRT_RET(
567 : ret != HCCL_SUCCESS,
568 : HCCL_ERROR(
569 : "[%s] thread resource get info failed. thread[0x%llx], resType[%d], infoLen[%u], info[%p]", __func__,
570 : static_cast<unsigned long long>(thread), static_cast<int32_t>(resType), infoLen, info),
571 : ret);
572 1 : HCCL_INFO("[%s]:comm[%s] get thread resource success.", __func__, commId.c_str());
573 1 : return HCCL_SUCCESS;
574 4 : }
|
