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 "thread.h"
12 : #include "cpu_ts_thread.h"
13 : #include "aicpu_ts_thread.h"
14 : #include "sal_pub.h"
15 : #include "stream_lite.h"
16 : #include "task_info.h"
17 : #include "comm_engine_utils.h"
18 : #include "aicpu_launch_manager.h"
19 : #include "dfx_profiling_handler_lite.h"
20 : #include "aicpu_indop_env.h"
21 : #include "adapter_rts_common.h"
22 :
23 : using namespace std;
24 :
25 : namespace hccl {
26 :
27 : struct DeviceThreadKey {
28 : int32_t deviceId;
29 : ThreadHandle handle;
30 :
31 16 : bool operator==(const DeviceThreadKey& other) const { return deviceId == other.deviceId && handle == other.handle; }
32 : };
33 :
34 : struct DeviceThreadKeyHash {
35 50 : std::size_t operator()(const DeviceThreadKey& key) const
36 : {
37 50 : return std::hash<int32_t>()(key.deviceId) ^ (std::hash<ThreadHandle>()(key.handle) << 1);
38 : }
39 : };
40 :
41 : static unordered_map<ThreadHandle, shared_ptr<Thread>> g_ThreadMap;
42 : static unordered_map<DeviceThreadKey, ThreadHandle, DeviceThreadKeyHash> g_ThreadD2HMap;
43 : static mutex g_ThreadMapMtx;
44 :
45 48 : HcclResult CreateThread(
46 : CommEngine engine, StreamType streamType, uint32_t notifyNum, NotifyLoadType loadType,
47 : shared_ptr<Thread>& out_thread)
48 : {
49 48 : out_thread = nullptr; // 初始化出参
50 :
51 48 : if (engine == COMM_ENGINE_CPU_TS || engine == COMM_ENGINE_CPU || engine == COMM_ENGINE_CCU
52 9 : || engine == COMM_ENGINE_AIV) {
53 39 : EXCEPTION_CATCH(out_thread = make_shared<CpuTsThread>(streamType, notifyNum, loadType), return HCCL_E_PTR);
54 48 : } else if (engine == COMM_ENGINE_AICPU_TS || engine == COMM_ENGINE_AICPU) {
55 8 : EXCEPTION_CATCH(out_thread = make_shared<AicpuTsThread>(streamType, notifyNum, loadType), return HCCL_E_PTR);
56 8 : } else {
57 1 : return HCCL_E_NOT_SUPPORT;
58 : }
59 :
60 47 : return HCCL_SUCCESS;
61 : }
62 :
63 12 : HcclResult CommHostEngineToNotifyLoadType(CommEngine engine, NotifyLoadType& type)
64 : {
65 12 : switch (engine) {
66 11 : case COMM_ENGINE_CPU:
67 : case COMM_ENGINE_CPU_TS:
68 : case COMM_ENGINE_CCU:
69 11 : type = NotifyLoadType::HOST_NOTIFY;
70 11 : break;
71 1 : default:
72 1 : HCCL_ERROR(
73 : "[ThreadMgr] Unsupported comm engine type: %s",
74 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str());
75 1 : return HCCL_E_PARA;
76 : }
77 11 : return HCCL_SUCCESS;
78 : }
79 :
80 18 : HcclResult CommEngineToNotifyLoadType(CommEngine engine, NotifyLoadType& type)
81 : {
82 18 : switch (engine) {
83 15 : case COMM_ENGINE_CPU:
84 : case COMM_ENGINE_CPU_TS:
85 : case COMM_ENGINE_CCU:
86 15 : type = NotifyLoadType::HOST_NOTIFY;
87 15 : break;
88 2 : case COMM_ENGINE_AICPU:
89 : case COMM_ENGINE_AICPU_TS:
90 2 : type = NotifyLoadType::DEVICE_NOTIFY;
91 2 : break;
92 1 : default:
93 1 : HCCL_ERROR(
94 : "[ThreadMgr] Unknown comm engine type: %s",
95 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str());
96 1 : return HCCL_E_PARA;
97 : }
98 17 : return HCCL_SUCCESS;
99 : }
100 :
101 17 : HcclResult CommEngineToStreamType(CommEngine engine, StreamType& type)
102 : {
103 17 : switch (engine) {
104 15 : case COMM_ENGINE_CPU:
105 : case COMM_ENGINE_CPU_TS:
106 : case COMM_ENGINE_CCU:
107 15 : type = StreamType::STREAM_TYPE_ONLINE; // 单算子使用online，图模式使用offine
108 15 : break;
109 2 : case COMM_ENGINE_AICPU:
110 : case COMM_ENGINE_AICPU_TS:
111 2 : type = StreamType::STREAM_TYPE_DEVICE;
112 2 : break;
113 : // 暂不支持AIV
114 0 : case COMM_ENGINE_AIV:
115 : default:
116 0 : HCCL_ERROR(
117 : "[ThreadMgr] Unknown comm engine type: %s",
118 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str());
119 0 : return HCCL_E_PARA;
120 : }
121 17 : return HCCL_SUCCESS;
122 : }
123 :
124 : struct EnumPairHash {
125 : template <class T1, class T2>
126 420 : std::size_t operator()(const std::pair<T1, T2>& p) const
127 : {
128 420 : const std::size_t h1 = std::hash<T1>{}(p.first);
129 420 : const std::size_t h2 = std::hash<T2>{}(p.second);
130 420 : return h1 ^ (h2 << 1);
131 : }
132 : };
133 :
134 : const std::unordered_map<std::pair<CommEngine, ThreadType>, NotifyLoadType, EnumPairHash> NOTIFY_TYPE_CONVERT = {
135 : {{COMM_ENGINE_CPU, THREAD_TYPE_TS}, NotifyLoadType::HOST_NOTIFY},
136 : {{COMM_ENGINE_CCU, THREAD_TYPE_TS}, NotifyLoadType::HOST_NOTIFY},
137 : {{COMM_ENGINE_AIV, THREAD_TYPE_TS}, NotifyLoadType::HOST_NOTIFY},
138 : {{COMM_ENGINE_AICPU, THREAD_TYPE_TS}, NotifyLoadType::DEVICE_NOTIFY},
139 : };
140 :
141 : const std::unordered_map<std::pair<CommEngine, ThreadType>, StreamType, EnumPairHash> STREAM_TYPE_CONVERT = {
142 : {{COMM_ENGINE_CPU, THREAD_TYPE_TS}, StreamType::STREAM_TYPE_ONLINE},
143 : {{COMM_ENGINE_CCU, THREAD_TYPE_TS}, StreamType::STREAM_TYPE_ONLINE},
144 : {{COMM_ENGINE_AIV, THREAD_TYPE_TS}, StreamType::STREAM_TYPE_ONLINE},
145 : {{COMM_ENGINE_AICPU, THREAD_TYPE_TS}, StreamType::STREAM_TYPE_DEVICE},
146 : };
147 :
148 25 : HcclResult GetNotifyLoadType(CommEngine engine, ThreadType threadType, NotifyLoadType& type)
149 : {
150 25 : auto iter = NOTIFY_TYPE_CONVERT.find(std::make_pair(engine, threadType));
151 25 : if (iter == NOTIFY_TYPE_CONVERT.end()) {
152 0 : HCCL_ERROR("[GetNotifyLoadType] not support comm engine type: %d, thread type: %d", engine, threadType);
153 0 : return HCCL_E_PARA;
154 : }
155 25 : type = iter->second;
156 25 : return HCCL_SUCCESS;
157 : }
158 :
159 19 : HcclResult GetStreamType(CommEngine engine, ThreadType threadType, StreamType& type)
160 : {
161 19 : auto iter = STREAM_TYPE_CONVERT.find(std::make_pair(engine, threadType));
162 19 : if (iter == STREAM_TYPE_CONVERT.end()) {
163 0 : HCCL_ERROR("[GetStreamType] not support comm engine type: %d, thread type: %d", engine, threadType);
164 0 : return HCCL_E_PARA;
165 : }
166 19 : type = iter->second;
167 19 : return HCCL_SUCCESS;
168 : }
169 :
170 : #ifndef CCL_KERNEL_AICPU
171 21 : HcclResult ValidateThreadParams(uint32_t threadNum, uint32_t notifyNumPerThread)
172 : {
173 21 : if (threadNum == 0 || threadNum > HCOMM_THREADNUM_MAX_NUM) {
174 2 : HCCL_ERROR(
175 : "[%s] Validate thread params failed. ThreadNum %u, range (0, %u]", __func__, threadNum,
176 : HCOMM_THREADNUM_MAX_NUM);
177 2 : return HCCL_E_PARA;
178 : }
179 19 : if (notifyNumPerThread > HCOMM_NOTIFY_MAX_NUM) {
180 1 : HCCL_ERROR(
181 : "[%s] Validate thread params failed. notifyNumPerThread %u, range [0, %u]", __func__, notifyNumPerThread,
182 : HCOMM_NOTIFY_MAX_NUM);
183 1 : return HCCL_E_PARA;
184 : }
185 18 : return HCCL_SUCCESS;
186 : }
187 :
188 15 : HcclResult SaveThreads(const vector<shared_ptr<Thread>>& newThreads)
189 : {
190 15 : int32_t deviceId = 0;
191 15 : CHK_RET(hrtGetDevice(&deviceId));
192 :
193 15 : lock_guard<mutex> lock(g_ThreadMapMtx);
194 33 : for (const auto& threadPtr : newThreads) {
195 18 : ThreadHandle handle = reinterpret_cast<ThreadHandle>(threadPtr.get());
196 :
197 18 : if (g_ThreadMap.find(handle) != g_ThreadMap.end()) {
198 0 : HCCL_ERROR("[%s] thread handle already exists [0x%llx] in ThreadMap", __func__, handle);
199 0 : return HCCL_E_INTERNAL;
200 : }
201 18 : DeviceThreadKey key{deviceId, handle};
202 18 : if (g_ThreadD2HMap.find(key) != g_ThreadD2HMap.end()) {
203 0 : HCCL_ERROR(
204 : "[%s] thread handle already exists [0x%llx] in g_ThreadD2HMap, deviceId[%d]", __func__, handle,
205 : deviceId);
206 0 : return HCCL_E_INTERNAL;
207 : }
208 :
209 18 : g_ThreadMap.emplace(handle, threadPtr);
210 18 : g_ThreadD2HMap.emplace(key, handle);
211 : }
212 15 : return HCCL_SUCCESS;
213 15 : }
214 :
215 17 : HcclResult CreateAndInitThreads(const ThreadCreateParams& params, vector<shared_ptr<Thread>>& outThreads)
216 : {
217 17 : HCCL_INFO(
218 : "[%s] Creating threads with params: engine[%s], threadNum[%u], "
219 : "notifyNumPerThread[%u], notifyLoadType[%u], streamType[%u]",
220 : __func__, GetEnumToString(GetCommEngineStatusStrMap(), params.engine).c_str(), params.threadNum,
221 : params.notifyNumPerThread, static_cast<int32_t>(params.notifyLoadType),
222 : static_cast<int32_t>(params.streamType));
223 17 : outThreads.reserve(params.threadNum);
224 :
225 35 : for (uint32_t i = 0; i < params.threadNum; ++i) {
226 20 : shared_ptr<Thread> threadPtr;
227 : // 创建线程
228 40 : HcclResult ret = CreateThread(
229 20 : params.engine, params.streamType, params.notifyNumPerThread, params.notifyLoadType, threadPtr);
230 20 : CHK_PRT_RET(
231 : ret != HCCL_SUCCESS, HCCL_ERROR("[%s] Failed to create thread at index %u, error: %d", __func__, i, ret),
232 : ret);
233 :
234 : // 初始化线程
235 20 : ret = threadPtr->Init();
236 20 : CHK_PRT_RET(
237 : ret != HCCL_SUCCESS,
238 : HCCL_ERROR("[%s] Failed to initialize thread at index %u, error: %d", __func__, i, ret), ret);
239 :
240 : // 添加到输出列表
241 18 : outThreads.emplace_back(move(threadPtr));
242 20 : }
243 15 : HCCL_INFO("[%s] Successfully created and initialized %u threads", __func__, params.threadNum);
244 15 : return HCCL_SUCCESS;
245 : }
246 :
247 1 : HcclResult FillThreadD2HMap(ThreadHandle* deviceThreadHandles, ThreadHandle* hostThreadHandles, uint32_t listNum)
248 : {
249 1 : int32_t deviceId = 0;
250 1 : CHK_RET(hrtGetDevice(&deviceId));
251 :
252 1 : lock_guard<mutex> lock(g_ThreadMapMtx);
253 2 : for (uint32_t idx = 0; idx < listNum; idx++) {
254 1 : auto deviceThreadHandle = deviceThreadHandles[idx];
255 1 : auto hostThreadHandle = hostThreadHandles[idx];
256 1 : HCCL_INFO(
257 : "%s deviceId[%d], deviceThreadHandle[0x%llx], hostThreadHandle[0x%llx]", __func__, deviceId,
258 : deviceThreadHandle, hostThreadHandle);
259 1 : DeviceThreadKey key{deviceId, deviceThreadHandle};
260 1 : g_ThreadD2HMap.emplace(key, hostThreadHandle);
261 : }
262 :
263 1 : return HCCL_SUCCESS;
264 1 : }
265 :
266 15 : HcclResult StoreThreadHandles(
267 : vector<shared_ptr<Thread>>& newThreads, ThreadHandle* threads, CommEngine engine, aclrtBinHandle binHandle)
268 : {
269 15 : CHK_PTR_NULL(threads);
270 15 : if (engine == COMM_ENGINE_AICPU || engine == COMM_ENGINE_AICPU_TS) {
271 : // AICPU引擎处理逻辑
272 0 : unique_ptr<ThreadHandle[]> aicpuHandle;
273 0 : EXCEPTION_CATCH(aicpuHandle = make_unique<ThreadHandle[]>(newThreads.size()), return HCCL_E_PTR);
274 0 : CHK_PTR_NULL(binHandle);
275 0 : HcclResult ret = AicpuLaunchMgr::ThreadKernelLaunchForBase(newThreads, aicpuHandle, binHandle);
276 :
277 0 : CHK_PRT_RET(
278 : ret != HCCL_SUCCESS,
279 : HCCL_ERROR(
280 : "[StoreThreadHandles] AiCpuKernelLaunch failed, engine[%s], return[%d].",
281 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), ret),
282 : ret);
283 :
284 : // 保存并映射AICPU线程句柄
285 0 : for (size_t i = 0; i < newThreads.size(); ++i) {
286 0 : threads[i] = aicpuHandle[i];
287 0 : ThreadHandle hostHandle = reinterpret_cast<ThreadHandle>(newThreads[i].get());
288 0 : CHK_RET(FillThreadD2HMap(&aicpuHandle[i], &hostHandle, 1));
289 0 : HCCL_INFO(
290 : "[StoreThreadHandles] AICPU engine[%s] threadArray[%zu] = [%lu]",
291 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), i, threads[i]);
292 : }
293 0 : } else {
294 33 : for (size_t i = 0; i < newThreads.size(); ++i) {
295 18 : threads[i] = reinterpret_cast<ThreadHandle>(newThreads[i].get());
296 18 : HCCL_INFO(
297 : "[StoreThreadHandles] Host engine[%s] threadArray[%zu] = [%lu]",
298 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), i, threads[i]);
299 : }
300 : }
301 15 : return HCCL_SUCCESS;
302 : }
303 :
304 : static HcclResult
305 22 : FreeThreadHandlesLocked(const ThreadHandle* threads, uint32_t threadNum, vector<ThreadHandle>& deviceHandles)
306 : {
307 22 : int32_t deviceId = 0;
308 22 : CHK_RET(hrtGetDevice(&deviceId));
309 :
310 22 : lock_guard<mutex> lock(g_ThreadMapMtx);
311 38 : for (uint32_t i = 0; i < threadNum; ++i) {
312 24 : const ThreadHandle inHandle = threads[i];
313 :
314 24 : DeviceThreadKey key{deviceId, inHandle};
315 24 : auto itH = g_ThreadD2HMap.find(key);
316 24 : if (itH == g_ThreadD2HMap.end()) {
317 8 : HCCL_ERROR(
318 : "[%s] failed to find handle mapping in g_ThreadD2HMap, deviceId[%d], inHandle[0x%llx].", __func__,
319 : deviceId, inHandle);
320 8 : return HcclResult::HCCL_E_NOT_FOUND;
321 : }
322 16 : const ThreadHandle mappedHandle = itH->second;
323 :
324 16 : auto itC = g_ThreadMap.find(mappedHandle);
325 16 : if (itC == g_ThreadMap.end()) {
326 0 : HCCL_ERROR(
327 : "[%s] failed to find thread in g_ThreadMap, deviceId[%d], inHandle[0x%llx], mappedHandle[0x%llx].",
328 : __func__, deviceId, inHandle, mappedHandle);
329 0 : return HcclResult::HCCL_E_NOT_FOUND;
330 : }
331 16 : if (inHandle != mappedHandle) {
332 0 : deviceHandles.push_back(inHandle);
333 : }
334 :
335 16 : HCCL_INFO(
336 : "[%s] erase thread: deviceId[%d], inHandle[0x%llx], mappedHandle[0x%llx], ptr[%p]", __func__, deviceId,
337 : inHandle, mappedHandle, itC->second.get());
338 16 : g_ThreadMap.erase(itC);
339 :
340 60 : for (auto it = g_ThreadD2HMap.begin(); it != g_ThreadD2HMap.end();) {
341 44 : if (it->second == mappedHandle && it->first.deviceId == deviceId) {
342 17 : it = g_ThreadD2HMap.erase(it);
343 : } else {
344 27 : ++it;
345 : }
346 : }
347 : }
348 14 : return HCCL_SUCCESS;
349 22 : }
350 :
351 23 : HcclResult FreeThreads(const ThreadHandle* threads, uint32_t threadNum, aclrtBinHandle binHandle)
352 : {
353 23 : CHK_PRT_RET(threads == nullptr, HCCL_ERROR("[HcommThreadfree] threads is null."), HCCL_E_PARA);
354 23 : if (threadNum == 0 || threadNum > HCOMM_THREADNUM_MAX_NUM) {
355 1 : HCCL_ERROR(
356 : "[%s] Validate thread params failed. ThreadNum %u, range (0, %u]", __func__, threadNum,
357 : HCOMM_THREADNUM_MAX_NUM);
358 1 : return HCCL_E_PARA;
359 : }
360 22 : HCCL_INFO("[%s] begin to free %u threads", __func__, threadNum);
361 :
362 22 : vector<ThreadHandle> deviceHandles; // 存放device侧的handle
363 :
364 22 : CHK_RET(FreeThreadHandlesLocked(threads, threadNum, deviceHandles));
365 :
366 : // 如果有需要销毁的deviceThread，调用销毁kernel
367 14 : if (!deviceHandles.empty()) {
368 0 : CHK_RET(AicpuLaunchMgr::ThreadKernelLaunchDestroy(deviceHandles.data(), deviceHandles.size(), binHandle));
369 : }
370 14 : HCCL_INFO("[%s] %u threads freed successfully.", __func__, threadNum);
371 14 : return HCCL_SUCCESS;
372 22 : }
373 :
374 4 : HcclResult SupplementThreadNotify(ThreadHandle handle, uint32_t notifyNum)
375 : {
376 4 : lock_guard<mutex> lock(g_ThreadMapMtx);
377 4 : auto it = g_ThreadMap.find(handle);
378 4 : CHK_PRT_RET(
379 : it == g_ThreadMap.end(), HCCL_ERROR("[%s] thread handle[0x%llx] not found in g_ThreadMap.", __func__, handle),
380 : HCCL_E_NOT_FOUND);
381 4 : if (it->second->GetNotifyNum() >= notifyNum) {
382 1 : return HCCL_SUCCESS;
383 : }
384 3 : u32 supplementNum = notifyNum - it->second->GetNotifyNum();
385 3 : HCCL_INFO(
386 : "[%s] supplement notify num:[%u], current notify num:[%u], target notify num:[%u]", __func__, supplementNum,
387 : it->second->GetNotifyNum(), notifyNum);
388 3 : return it->second->SupplementNotify(supplementNum);
389 4 : }
390 :
391 2 : HcclResult LookupThreadByHandle(ThreadHandle handle, std::shared_ptr<Thread>& outThread)
392 : {
393 2 : lock_guard<mutex> lock(g_ThreadMapMtx);
394 2 : auto it = g_ThreadMap.find(handle);
395 :
396 2 : if (it == g_ThreadMap.end()) {
397 : // try find device handle
398 0 : int32_t deviceId = 0;
399 0 : CHK_RET(hrtGetDevice(&deviceId));
400 0 : DeviceThreadKey key{deviceId, handle};
401 0 : auto device_it = g_ThreadD2HMap.find(key);
402 0 : CHK_PRT_RET(
403 : device_it == g_ThreadD2HMap.end(),
404 : HCCL_ERROR(
405 : "[%s] device handle[0x%llx] not found in g_ThreadD2HMap, deviceId[%d].", __func__, handle, deviceId),
406 : HCCL_E_NOT_FOUND);
407 0 : it = g_ThreadMap.find(device_it->second);
408 0 : CHK_PRT_RET(
409 : it == g_ThreadMap.end(),
410 : HCCL_ERROR("[%s] thread handle[0x%llx] not found in g_ThreadMap.", __func__, handle), HCCL_E_NOT_FOUND);
411 : }
412 :
413 2 : outThread = it->second;
414 2 : return HCCL_SUCCESS;
415 2 : }
416 :
417 0 : HcclResult LookupD2HHandle(ThreadHandle deviceHandle, ThreadHandle& outHostHandle)
418 : {
419 0 : int32_t deviceId = 0;
420 0 : CHK_RET(hrtGetDevice(&deviceId));
421 0 : lock_guard<mutex> lock(g_ThreadMapMtx);
422 0 : DeviceThreadKey key{deviceId, deviceHandle};
423 0 : auto it = g_ThreadD2HMap.find(key);
424 0 : CHK_PRT_RET(
425 : it == g_ThreadD2HMap.end(),
426 : HCCL_ERROR(
427 : "[%s] device handle[0x%llx] not found in g_ThreadD2HMap, deviceId[%d].", __func__, deviceHandle, deviceId),
428 : HCCL_E_NOT_FOUND);
429 0 : outHostHandle = it->second;
430 0 : return HCCL_SUCCESS;
431 0 : }
432 : #endif
433 :
434 8 : HcclResult Thread::AddThreadHandleToMap(CommEngine commEngine, ThreadHandle threadHandle)
435 : {
436 8 : if (threadHandleMap_.find(commEngine) != threadHandleMap_.end() && threadHandleMap_[commEngine] != threadHandle) {
437 0 : HCCL_ERROR(
438 : "[Thread][%s]Mapping already exists:commEngine[%s], threadHandle[%lu], new threadHandle[%lu]", __func__,
439 : GetEnumToString(GetCommEngineStatusStrMap(), commEngine).c_str(), threadHandleMap_[commEngine],
440 : threadHandle);
441 : }
442 :
443 8 : threadHandleMap_[commEngine] = threadHandle;
444 8 : return HCCL_SUCCESS;
445 : }
446 :
447 3 : Thread* Thread::FindThreadByCommEngine(CommEngine commEngine)
448 : {
449 3 : if (threadHandleMap_.find(commEngine) != threadHandleMap_.end()) {
450 0 : return reinterpret_cast<Thread*>(threadHandleMap_[commEngine]);
451 : }
452 :
453 3 : return nullptr;
454 : }
455 :
456 0 : HcclResult Thread::ReportAicpuNotifyWaitTask(u64 notifyId, u64 beginTime, u32 taskId, u32 sqId) const
457 : {
458 0 : if (!IsReportTask()) {
459 0 : return HCCL_SUCCESS;
460 : }
461 0 : Hccl::TaskParam taskParam{};
462 0 : taskParam.taskType = Hccl::TaskParamType::TASK_NOTIFY_WAIT;
463 0 : taskParam.beginTime = beginTime;
464 0 : taskParam.taskPara.Notify.notifyID = notifyId;
465 0 : taskParam.taskPara.Notify.value = 1;
466 0 : taskParam.endTime = ProfGetCurCpuTimestamp();
467 0 : CHK_PTR_NULL(callback_);
468 0 : CHK_RET(callback_(sqId, taskId, taskParam, DFX_INVALID_U64));
469 0 : HCCL_INFO(
470 : "[Thread][%s] sqId[%u], taskId[%u], notifyId[%llu], %s", __func__, sqId, taskId, notifyId,
471 : taskParam.Describe().c_str());
472 0 : return HCCL_SUCCESS;
473 0 : }
474 :
475 1 : HcclResult Thread::ReportHostNotifyWaitTask(
476 : [[maybe_unused]] u64 notifyId, [[maybe_unused]] u64 beginTime, [[maybe_unused]] bool isMaster) const
477 : {
478 : #ifndef CCL_KERNEL_AICPU
479 1 : Hccl::TaskParam taskParam{};
480 1 : taskParam.taskType = Hccl::TaskParamType::TASK_NOTIFY_WAIT;
481 1 : taskParam.beginTime = beginTime;
482 1 : taskParam.taskPara.Notify.notifyID = notifyId;
483 1 : taskParam.taskPara.Notify.value = 1;
484 1 : taskParam.isMaster = isMaster;
485 1 : u32 taskId = 0;
486 1 : u32 streamId = 0;
487 1 : hrtGetTaskIdAndStreamID(taskId, streamId);
488 1 : taskParam.endTime = Hccl::DlProfFunction::GetInstance().dlMsprofSysCycleTime();
489 1 : HCCL_INFO("[ReportHostNotifyWaitTask] time is %llu", taskParam.endTime);
490 1 : CHK_PTR_NULL(callback_);
491 1 : CHK_RET(callback_(streamId, taskId, taskParam, DFX_INVALID_U64));
492 1 : HCCL_INFO(
493 : "[Thread][%s] streamId[%u], taskId[%u], notifyId[%llu], %s", __func__, streamId, taskId, notifyId,
494 : taskParam.Describe().c_str());
495 : #endif
496 1 : return HCCL_SUCCESS;
497 1 : }
498 :
499 0 : HcclResult Thread::ReportAicpuNotifyRecordTask(u64 notifyId, u64 beginTime, u32 taskId, u32 sqId) const
500 : {
501 0 : if (!IsReportTask()) {
502 0 : return HCCL_SUCCESS;
503 : }
504 0 : Hccl::TaskParam taskParam{};
505 0 : taskParam.taskType = Hccl::TaskParamType::TASK_NOTIFY_RECORD;
506 0 : taskParam.beginTime = beginTime;
507 0 : taskParam.taskPara.Notify.notifyID = notifyId;
508 0 : taskParam.taskPara.Notify.value = 1;
509 0 : taskParam.endTime = ProfGetCurCpuTimestamp();
510 0 : CHK_PTR_NULL(callback_);
511 0 : CHK_RET(callback_(sqId, taskId, taskParam, DFX_INVALID_U64));
512 0 : HCCL_INFO(
513 : "[Thread][%s] sqId[%u], taskId[%u], notifyId[%llu], %s", __func__, sqId, taskId, notifyId,
514 : taskParam.Describe().c_str());
515 0 : return HCCL_SUCCESS;
516 0 : }
517 :
518 1 : HcclResult Thread::ReportHostNotifyRecordTask(
519 : [[maybe_unused]] u64 notifyId, [[maybe_unused]] u64 beginTime, [[maybe_unused]] bool isMaster) const
520 : {
521 : #ifndef CCL_KERNEL_AICPU
522 1 : Hccl::TaskParam taskParam{};
523 1 : taskParam.taskType = Hccl::TaskParamType::TASK_NOTIFY_RECORD;
524 1 : taskParam.beginTime = beginTime;
525 1 : taskParam.taskPara.Notify.notifyID = notifyId;
526 1 : taskParam.taskPara.Notify.value = 1;
527 1 : taskParam.isMaster = isMaster;
528 1 : u32 taskId = 0;
529 1 : u32 streamId = 0;
530 1 : hrtGetTaskIdAndStreamID(taskId, streamId);
531 1 : taskParam.endTime = Hccl::DlProfFunction::GetInstance().dlMsprofSysCycleTime();
532 1 : HCCL_INFO("[ReportHostNotifyRecordTask] time is %llu", taskParam.endTime);
533 1 : CHK_PTR_NULL(callback_);
534 1 : CHK_RET(callback_(streamId, taskId, taskParam, DFX_INVALID_U64));
535 1 : HCCL_INFO(
536 : "[Thread][%s] streamId[%u], taskId[%u], notifyId[%llu], %s", __func__, streamId, taskId, notifyId,
537 : taskParam.Describe().c_str());
538 : #endif
539 1 : return HCCL_SUCCESS;
540 1 : }
541 :
542 1 : HcclResult Thread::ReportHostLocalCopyTask(
543 : [[maybe_unused]] void* dst, [[maybe_unused]] const void* src, [[maybe_unused]] uint64_t sizeByte,
544 : [[maybe_unused]] u64 beginTime, [[maybe_unused]] bool isMaster) const
545 : {
546 : #ifndef CCL_KERNEL_AICPU
547 1 : Hccl::TaskParam taskParam{};
548 1 : taskParam.taskType = Hccl::TaskParamType::TASK_SDMA;
549 1 : taskParam.beginTime = beginTime;
550 1 : taskParam.taskPara.DMA.src = src;
551 1 : taskParam.taskPara.DMA.dst = dst;
552 1 : taskParam.taskPara.DMA.size = sizeByte;
553 1 : taskParam.taskPara.DMA.notifyID = DFX_INVALID_U64;
554 1 : taskParam.taskPara.DMA.linkType = Hccl::DfxLinkType::ONCHIP;
555 1 : taskParam.taskPara.DMA.dmaOp = Hccl::DmaOp::HCCL_DMA_READ;
556 1 : taskParam.isMaster = isMaster;
557 :
558 1 : u32 taskId = 0;
559 1 : u32 streamId = 0;
560 1 : hrtGetTaskIdAndStreamID(taskId, streamId);
561 1 : taskParam.endTime = Hccl::DlProfFunction::GetInstance().dlMsprofSysCycleTime();
562 1 : CHK_PTR_NULL(callback_);
563 1 : CHK_RET(callback_(streamId, taskId, taskParam, DFX_INVALID_U64));
564 1 : HCCL_INFO(
565 : "[Thread][%s] streamId[%u], taskId[%u], src[%p], dst[%p], len[%llu] %s", __func__, streamId, taskId, src, dst,
566 : sizeByte, taskParam.Describe().c_str());
567 : #endif
568 1 : return HCCL_SUCCESS;
569 1 : }
570 :
571 0 : HcclResult Thread::ReportAicpuLocalCopyTask(
572 : void* dst, const void* src, uint64_t sizeByte, u64 beginTime, u32 taskId, u32 sqId) const
573 : {
574 0 : if (!IsReportTask()) {
575 0 : return HCCL_SUCCESS;
576 : }
577 0 : Hccl::TaskParam taskParam{};
578 0 : taskParam.taskType = Hccl::TaskParamType::TASK_SDMA;
579 0 : taskParam.beginTime = beginTime;
580 0 : taskParam.taskPara.DMA.src = src;
581 0 : taskParam.taskPara.DMA.dst = dst;
582 0 : taskParam.taskPara.DMA.size = sizeByte;
583 0 : taskParam.taskPara.DMA.notifyID = DFX_INVALID_U64;
584 0 : taskParam.taskPara.DMA.linkType = Hccl::DfxLinkType::ONCHIP;
585 0 : taskParam.taskPara.DMA.dmaOp = Hccl::DmaOp::HCCL_DMA_READ;
586 0 : taskParam.endTime = ProfGetCurCpuTimestamp();
587 0 : CHK_PTR_NULL(callback_);
588 0 : CHK_RET(callback_(sqId, taskId, taskParam, DFX_INVALID_U64));
589 0 : HCCL_INFO(
590 : "[Thread][%s] sqId[%u], taskId[%u], src[%p], dst[%p], len[%llu] %s", __func__, sqId, taskId, src, dst, sizeByte,
591 : taskParam.Describe().c_str());
592 0 : return HCCL_SUCCESS;
593 0 : }
594 :
595 0 : HcclResult Thread::ReportAicpuLocalReduceTask(
596 : void* dst, const void* src, uint64_t sizeByte, HcommDataType dataType, HcommReduceOp reduceOp, u64 beginTime,
597 : u32 taskId, u32 sqId) const
598 : {
599 0 : if (!IsReportTask()) {
600 0 : return HCCL_SUCCESS;
601 : }
602 0 : Hccl::TaskParam taskParam{};
603 0 : taskParam.taskType = Hccl::TaskParamType::TASK_REDUCE_INLINE;
604 0 : taskParam.beginTime = beginTime;
605 0 : taskParam.taskPara.Reduce.src = src;
606 0 : taskParam.taskPara.Reduce.dst = dst;
607 0 : taskParam.taskPara.Reduce.size = sizeByte;
608 0 : taskParam.taskPara.Reduce.notifyID = DFX_INVALID_U64;
609 0 : taskParam.taskPara.Reduce.linkType = Hccl::DfxLinkType::ONCHIP;
610 0 : taskParam.taskPara.Reduce.dataType = static_cast<HcclDataType>(dataType);
611 0 : taskParam.taskPara.Reduce.reduceOp = static_cast<HcclReduceOp>(reduceOp);
612 0 : CHK_PTR_NULL(callback_);
613 0 : CHK_RET(callback_(sqId, taskId, taskParam, DFX_INVALID_U64));
614 0 : HCCL_INFO(
615 : "[Thread][%s] sqId[%u], taskId[%u], src[%p], dst[%p], len[%llu], dataType[%d], reduceOp[%d], %s", __func__,
616 : sqId, taskId, src, dst, sizeByte, dataType, reduceOp, taskParam.Describe().c_str());
617 0 : return HCCL_SUCCESS;
618 0 : }
619 :
620 1 : HcclResult Thread::ReportHostLocalReduceTask(
621 : [[maybe_unused]] void* dst, [[maybe_unused]] const void* src, [[maybe_unused]] uint64_t sizeByte,
622 : [[maybe_unused]] HcommDataType dataType, [[maybe_unused]] HcommReduceOp reduceOp, [[maybe_unused]] u64 beginTime,
623 : [[maybe_unused]] bool isMaster) const
624 : {
625 : #ifndef CCL_KERNEL_AICPU
626 1 : Hccl::TaskParam taskParam{};
627 1 : taskParam.taskType = Hccl::TaskParamType::TASK_REDUCE_INLINE;
628 1 : taskParam.beginTime = beginTime;
629 1 : taskParam.taskPara.Reduce.src = src;
630 1 : taskParam.taskPara.Reduce.dst = dst;
631 1 : taskParam.taskPara.Reduce.size = sizeByte;
632 1 : taskParam.taskPara.Reduce.notifyID = DFX_INVALID_U64;
633 1 : taskParam.taskPara.Reduce.linkType = Hccl::DfxLinkType::ONCHIP;
634 1 : taskParam.taskPara.Reduce.dataType = static_cast<HcclDataType>(dataType);
635 1 : taskParam.taskPara.Reduce.reduceOp = static_cast<HcclReduceOp>(reduceOp);
636 1 : taskParam.isMaster = isMaster;
637 1 : u32 taskId = 0;
638 1 : u32 streamId = 0;
639 1 : hrtGetTaskIdAndStreamID(taskId, streamId);
640 1 : taskParam.endTime = Hccl::DlProfFunction::GetInstance().dlMsprofSysCycleTime();
641 :
642 1 : CHK_PTR_NULL(callback_);
643 1 : CHK_RET(callback_(streamId, taskId, taskParam, DFX_INVALID_U64));
644 1 : HCCL_INFO(
645 : "[Thread][%s] streamId[%u], taskId[%u], src[%p], dst[%p], len[%llu], dataType[%d], reduceOp[%d] %s", __func__,
646 : streamId, taskId, src, dst, sizeByte, dataType, reduceOp, taskParam.Describe().c_str());
647 : #endif
648 1 : return HCCL_SUCCESS;
649 1 : }
650 :
651 0 : bool Thread::IsReportTask() const
652 : {
653 : #ifdef CCL_KERNEL_AICPU
654 : return hcomm::GetTaskExceptionEnable() || Hccl::DfxProfilingHandlerLite::GetInstance().GetProfL1State();
655 : #endif
656 0 : return true;
657 : }
658 :
659 : } // namespace hccl
|
