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