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