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 460 : std::size_t operator()(const std::pair<T1, T2>& p) const
127 : {
128 460 : const std::size_t h1 = std::hash<T1>{}(p.first);
129 460 : const std::size_t h2 = std::hash<T2>{}(p.second);
130 460 : 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 0 : return HCCL_E_PARA;
442 : }
443 :
444 8 : threadHandleMap_[commEngine] = threadHandle;
445 8 : return HCCL_SUCCESS;
446 : }
447 :
448 3 : Thread* Thread::FindThreadByCommEngine(CommEngine commEngine)
449 : {
450 3 : if (threadHandleMap_.find(commEngine) != threadHandleMap_.end()) {
451 0 : return reinterpret_cast<Thread*>(threadHandleMap_[commEngine]);
452 : }
453 :
454 3 : return nullptr;
455 : }
456 :
457 0 : HcclResult Thread::ReportAicpuNotifyWaitTask(u64 notifyId, u64 beginTime, u32 taskId, u32 sqId) const
458 : {
459 0 : if (!IsReportTask()) {
460 0 : return HCCL_SUCCESS;
461 : }
462 0 : Hccl::TaskParam taskParam{};
463 0 : taskParam.taskType = Hccl::TaskParamType::TASK_NOTIFY_WAIT;
464 0 : taskParam.beginTime = beginTime;
465 0 : taskParam.taskPara.Notify.notifyID = notifyId;
466 0 : taskParam.taskPara.Notify.value = 1;
467 0 : taskParam.endTime = ProfGetCurCpuTimestamp();
468 0 : CHK_PTR_NULL(callback_);
469 0 : CHK_RET(callback_(sqId, taskId, taskParam, DFX_INVALID_U64));
470 0 : HCCL_INFO(
471 : "[Thread][%s] sqId[%u], taskId[%u], notifyId[%llu], %s", __func__, sqId, taskId, notifyId,
472 : taskParam.Describe().c_str());
473 0 : return HCCL_SUCCESS;
474 0 : }
475 :
476 1 : HcclResult Thread::ReportHostNotifyWaitTask(
477 : [[maybe_unused]] u64 notifyId, [[maybe_unused]] u64 beginTime, [[maybe_unused]] bool isMaster) const
478 : {
479 : #ifndef CCL_KERNEL_AICPU
480 1 : Hccl::TaskParam taskParam{};
481 1 : taskParam.taskType = Hccl::TaskParamType::TASK_NOTIFY_WAIT;
482 1 : taskParam.beginTime = beginTime;
483 1 : taskParam.taskPara.Notify.notifyID = notifyId;
484 1 : taskParam.taskPara.Notify.value = 1;
485 1 : taskParam.isMaster = isMaster;
486 1 : u32 taskId = 0;
487 1 : u32 streamId = 0;
488 1 : hrtGetTaskIdAndStreamID(taskId, streamId);
489 1 : taskParam.endTime = Hccl::DlProfFunction::GetInstance().dlMsprofSysCycleTime();
490 1 : HCCL_INFO("[ReportHostNotifyWaitTask] time is %llu", taskParam.endTime);
491 1 : CHK_PTR_NULL(callback_);
492 1 : CHK_RET(callback_(streamId, taskId, taskParam, DFX_INVALID_U64));
493 1 : HCCL_INFO(
494 : "[Thread][%s] streamId[%u], taskId[%u], notifyId[%llu], %s", __func__, streamId, taskId, notifyId,
495 : taskParam.Describe().c_str());
496 : #endif
497 1 : return HCCL_SUCCESS;
498 1 : }
499 :
500 0 : HcclResult Thread::ReportAicpuNotifyRecordTask(u64 notifyId, u64 beginTime, u32 taskId, u32 sqId) const
501 : {
502 0 : if (!IsReportTask()) {
503 0 : return HCCL_SUCCESS;
504 : }
505 0 : Hccl::TaskParam taskParam{};
506 0 : taskParam.taskType = Hccl::TaskParamType::TASK_NOTIFY_RECORD;
507 0 : taskParam.beginTime = beginTime;
508 0 : taskParam.taskPara.Notify.notifyID = notifyId;
509 0 : taskParam.taskPara.Notify.value = 1;
510 0 : taskParam.endTime = ProfGetCurCpuTimestamp();
511 0 : CHK_PTR_NULL(callback_);
512 0 : CHK_RET(callback_(sqId, taskId, taskParam, DFX_INVALID_U64));
513 0 : HCCL_INFO(
514 : "[Thread][%s] sqId[%u], taskId[%u], notifyId[%llu], %s", __func__, sqId, taskId, notifyId,
515 : taskParam.Describe().c_str());
516 0 : return HCCL_SUCCESS;
517 0 : }
518 :
519 1 : HcclResult Thread::ReportHostNotifyRecordTask(
520 : [[maybe_unused]] u64 notifyId, [[maybe_unused]] u64 beginTime, [[maybe_unused]] bool isMaster) const
521 : {
522 : #ifndef CCL_KERNEL_AICPU
523 1 : Hccl::TaskParam taskParam{};
524 1 : taskParam.taskType = Hccl::TaskParamType::TASK_NOTIFY_RECORD;
525 1 : taskParam.beginTime = beginTime;
526 1 : taskParam.taskPara.Notify.notifyID = notifyId;
527 1 : taskParam.taskPara.Notify.value = 1;
528 1 : taskParam.isMaster = isMaster;
529 1 : u32 taskId = 0;
530 1 : u32 streamId = 0;
531 1 : hrtGetTaskIdAndStreamID(taskId, streamId);
532 1 : taskParam.endTime = Hccl::DlProfFunction::GetInstance().dlMsprofSysCycleTime();
533 1 : HCCL_INFO("[ReportHostNotifyRecordTask] time is %llu", taskParam.endTime);
534 1 : CHK_PTR_NULL(callback_);
535 1 : CHK_RET(callback_(streamId, taskId, taskParam, DFX_INVALID_U64));
536 1 : HCCL_INFO(
537 : "[Thread][%s] streamId[%u], taskId[%u], notifyId[%llu], %s", __func__, streamId, taskId, notifyId,
538 : taskParam.Describe().c_str());
539 : #endif
540 1 : return HCCL_SUCCESS;
541 1 : }
542 :
543 1 : HcclResult Thread::ReportHostLocalCopyTask(
544 : [[maybe_unused]] void* dst, [[maybe_unused]] const void* src, [[maybe_unused]] uint64_t sizeByte,
545 : [[maybe_unused]] u64 beginTime, [[maybe_unused]] bool isMaster) const
546 : {
547 : #ifndef CCL_KERNEL_AICPU
548 1 : Hccl::TaskParam taskParam{};
549 1 : taskParam.taskType = Hccl::TaskParamType::TASK_SDMA;
550 1 : taskParam.beginTime = beginTime;
551 1 : taskParam.taskPara.DMA.src = src;
552 1 : taskParam.taskPara.DMA.dst = dst;
553 1 : taskParam.taskPara.DMA.size = sizeByte;
554 1 : taskParam.taskPara.DMA.notifyID = DFX_INVALID_U64;
555 1 : taskParam.taskPara.DMA.linkType = Hccl::DfxLinkType::ONCHIP;
556 1 : taskParam.taskPara.DMA.dmaOp = Hccl::DmaOp::HCCL_DMA_READ;
557 1 : taskParam.isMaster = isMaster;
558 :
559 1 : u32 taskId = 0;
560 1 : u32 streamId = 0;
561 1 : hrtGetTaskIdAndStreamID(taskId, streamId);
562 1 : taskParam.endTime = Hccl::DlProfFunction::GetInstance().dlMsprofSysCycleTime();
563 1 : CHK_PTR_NULL(callback_);
564 1 : CHK_RET(callback_(streamId, taskId, taskParam, DFX_INVALID_U64));
565 1 : HCCL_INFO(
566 : "[Thread][%s] streamId[%u], taskId[%u], src[%p], dst[%p], len[%llu] %s", __func__, streamId, taskId, src, dst,
567 : sizeByte, taskParam.Describe().c_str());
568 : #endif
569 1 : return HCCL_SUCCESS;
570 1 : }
571 :
572 0 : HcclResult Thread::ReportAicpuLocalCopyTask(
573 : void* dst, const void* src, uint64_t sizeByte, u64 beginTime, u32 taskId, u32 sqId) const
574 : {
575 0 : if (!IsReportTask()) {
576 0 : return HCCL_SUCCESS;
577 : }
578 0 : Hccl::TaskParam taskParam{};
579 0 : taskParam.taskType = Hccl::TaskParamType::TASK_SDMA;
580 0 : taskParam.beginTime = beginTime;
581 0 : taskParam.taskPara.DMA.src = src;
582 0 : taskParam.taskPara.DMA.dst = dst;
583 0 : taskParam.taskPara.DMA.size = sizeByte;
584 0 : taskParam.taskPara.DMA.notifyID = DFX_INVALID_U64;
585 0 : taskParam.taskPara.DMA.linkType = Hccl::DfxLinkType::ONCHIP;
586 0 : taskParam.taskPara.DMA.dmaOp = Hccl::DmaOp::HCCL_DMA_READ;
587 0 : taskParam.endTime = ProfGetCurCpuTimestamp();
588 0 : CHK_PTR_NULL(callback_);
589 0 : CHK_RET(callback_(sqId, taskId, taskParam, DFX_INVALID_U64));
590 0 : HCCL_INFO(
591 : "[Thread][%s] sqId[%u], taskId[%u], src[%p], dst[%p], len[%llu] %s", __func__, sqId, taskId, src, dst, sizeByte,
592 : taskParam.Describe().c_str());
593 0 : return HCCL_SUCCESS;
594 0 : }
595 :
596 0 : HcclResult Thread::ReportAicpuLocalReduceTask(
597 : void* dst, const void* src, uint64_t sizeByte, HcommDataType dataType, HcommReduceOp reduceOp, u64 beginTime,
598 : u32 taskId, u32 sqId) const
599 : {
600 0 : if (!IsReportTask()) {
601 0 : return HCCL_SUCCESS;
602 : }
603 0 : Hccl::TaskParam taskParam{};
604 0 : taskParam.taskType = Hccl::TaskParamType::TASK_REDUCE_INLINE;
605 0 : taskParam.beginTime = beginTime;
606 0 : taskParam.taskPara.Reduce.src = src;
607 0 : taskParam.taskPara.Reduce.dst = dst;
608 0 : taskParam.taskPara.Reduce.size = sizeByte;
609 0 : taskParam.taskPara.Reduce.notifyID = DFX_INVALID_U64;
610 0 : taskParam.taskPara.Reduce.linkType = Hccl::DfxLinkType::ONCHIP;
611 0 : taskParam.taskPara.Reduce.dataType = static_cast<HcclDataType>(dataType);
612 0 : taskParam.taskPara.Reduce.reduceOp = static_cast<HcclReduceOp>(reduceOp);
613 0 : CHK_PTR_NULL(callback_);
614 0 : CHK_RET(callback_(sqId, taskId, taskParam, DFX_INVALID_U64));
615 0 : HCCL_INFO(
616 : "[Thread][%s] sqId[%u], taskId[%u], src[%p], dst[%p], len[%llu], dataType[%d], reduceOp[%d], %s", __func__,
617 : sqId, taskId, src, dst, sizeByte, dataType, reduceOp, taskParam.Describe().c_str());
618 0 : return HCCL_SUCCESS;
619 0 : }
620 :
621 1 : HcclResult Thread::ReportHostLocalReduceTask(
622 : [[maybe_unused]] void* dst, [[maybe_unused]] const void* src, [[maybe_unused]] uint64_t sizeByte,
623 : [[maybe_unused]] HcommDataType dataType, [[maybe_unused]] HcommReduceOp reduceOp, [[maybe_unused]] u64 beginTime,
624 : [[maybe_unused]] bool isMaster) const
625 : {
626 : #ifndef CCL_KERNEL_AICPU
627 1 : Hccl::TaskParam taskParam{};
628 1 : taskParam.taskType = Hccl::TaskParamType::TASK_REDUCE_INLINE;
629 1 : taskParam.beginTime = beginTime;
630 1 : taskParam.taskPara.Reduce.src = src;
631 1 : taskParam.taskPara.Reduce.dst = dst;
632 1 : taskParam.taskPara.Reduce.size = sizeByte;
633 1 : taskParam.taskPara.Reduce.notifyID = DFX_INVALID_U64;
634 1 : taskParam.taskPara.Reduce.linkType = Hccl::DfxLinkType::ONCHIP;
635 1 : taskParam.taskPara.Reduce.dataType = static_cast<HcclDataType>(dataType);
636 1 : taskParam.taskPara.Reduce.reduceOp = static_cast<HcclReduceOp>(reduceOp);
637 1 : taskParam.isMaster = isMaster;
638 1 : u32 taskId = 0;
639 1 : u32 streamId = 0;
640 1 : hrtGetTaskIdAndStreamID(taskId, streamId);
641 1 : taskParam.endTime = Hccl::DlProfFunction::GetInstance().dlMsprofSysCycleTime();
642 :
643 1 : CHK_PTR_NULL(callback_);
644 1 : CHK_RET(callback_(streamId, taskId, taskParam, DFX_INVALID_U64));
645 1 : HCCL_INFO(
646 : "[Thread][%s] streamId[%u], taskId[%u], src[%p], dst[%p], len[%llu], dataType[%d], reduceOp[%d] %s", __func__,
647 : streamId, taskId, src, dst, sizeByte, dataType, reduceOp, taskParam.Describe().c_str());
648 : #endif
649 1 : return HCCL_SUCCESS;
650 1 : }
651 :
652 0 : bool Thread::IsReportTask() const
653 : {
654 : #ifdef CCL_KERNEL_AICPU
655 : return hcomm::GetTaskExceptionEnable() || Hccl::DfxProfilingHandlerLite::GetInstance().GetProfL1State();
656 : #endif
657 0 : return true;
658 : }
659 :
660 : } // namespace hccl
|
