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