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 "hccl_api_data.h"
12 : #include "new/hccl_primitive_local.h"
13 : #include "new/hccl_primitive_remote.h"
14 : #include "thread.h"
15 : #include "launch_context.h"
16 :
17 : #include "ub_transport_lite_impl.h"
18 : #include "device/framework/aicpu_hccl_process.h"
19 : #include "coll_comm_aicpu_mgr.h"
20 : #include "aicpu_indop_process.h"
21 : #include "aicpu_indop_env.h"
22 : #include "profiling_handler_lite.h"
23 : #include "hcclCommOp.h"
24 : #include "hcomm_diag.h"
25 : #include "aicpu_ts_primitives_c_adpt.h"
26 : #include "hccl_diag.h"
27 : #include "channel.h"
28 : #include "aicpu_ts_channel_helper.h"
29 : #include "unified_platform/pub_inc/config_plf_log.h"
30 :
31 : using Hccl::PLF_DATA_OP;
32 :
33 : using namespace hccl;
34 : thread_local LaunchContext g_threadLaunchCtx;
35 :
36 0 : bool IsBatchLaunchMode() {
37 0 : return g_threadLaunchCtx.IsBatchLaunchMode();
38 : }
39 :
40 0 : uint32_t GetSqFullTimeOut() {
41 0 : return g_threadLaunchCtx.GetSqFullTimeOut();
42 : }
43 :
44 :
45 0 : inline bool GetProfilingEnable() {
46 0 : return Hccl::ProfilingHandlerLite::GetInstance().GetProfL0State() ||
47 0 : Hccl::ProfilingHandlerLite::GetInstance().GetProfL1State();
48 : }
49 :
50 61 : void AddThread(ThreadHandle thread) {
51 61 : g_threadLaunchCtx.AddThread(thread);
52 61 : }
53 :
54 1 : HcclResult HandleDispatchAllStreams()
55 : {
56 1 : return g_threadLaunchCtx.HandleDispatchAllStreams();
57 : }
58 :
59 0 : bool IsSupportReduce(HcommDataType dataType, HcommReduceOp op)
60 : {
61 0 : bool checkDataType =
62 0 : (dataType == HCOMM_DATA_TYPE_FP32 || dataType == HCOMM_DATA_TYPE_FP16 || dataType == HCOMM_DATA_TYPE_INT8 ||
63 0 : dataType == HCOMM_DATA_TYPE_INT16 || dataType == HCOMM_DATA_TYPE_INT32 || dataType == HCOMM_DATA_TYPE_BFP16);
64 0 : bool checkReduceType = (op == HCOMM_REDUCE_SUM || op == HCOMM_REDUCE_MAX || op == HCOMM_REDUCE_MIN);
65 0 : return checkDataType && checkReduceType;
66 : }
67 :
68 0 : HcclResult HcommThreadGetNotifyId(ThreadHandle thread, uint32_t notifyIdx, uint32_t *notifyId)
69 : {
70 0 : Thread *const threadPtr = reinterpret_cast<Thread *>(thread);
71 0 : CHK_PTR_NULL(threadPtr);
72 0 : LocalNotify *const notifyPtr = threadPtr->GetNotify(notifyIdx);
73 0 : CHK_PTR_NULL(notifyPtr);
74 0 : *notifyId = notifyPtr->notifyId_;
75 :
76 0 : return HCCL_SUCCESS;
77 : }
78 :
79 0 : HcclResult HcclDfxRegOpInfoByCommId(char* commId, void* hcclDfxOpInfo)
80 : {
81 0 : if (!GetProfilingEnable() && !hcomm::GetTaskExceptionEnable()) {
82 0 : return HCCL_SUCCESS;
83 : }
84 0 : CHK_PTR_NULL(commId);
85 0 : CHK_PTR_NULL(hcclDfxOpInfo);
86 :
87 : DevType deviceType;
88 0 : CHK_RET(hrtGetDeviceType(deviceType));
89 0 : if (deviceType == DevType::DEV_TYPE_910B) {
90 0 : HCCL_INFO("[%s]not support, comId[%s], devType[%d]", __func__, commId, deviceType);
91 0 : return HCCL_SUCCESS;
92 : }
93 :
94 0 : HcclDfxOpInfo *aicpuDfxInfo = reinterpret_cast<HcclDfxOpInfo *>(hcclDfxOpInfo);
95 0 : CHK_RET(HcommThreadGetNotifyId(aicpuDfxInfo->cpuTsThread, aicpuDfxInfo->cpuWaitAicpuNotifyIdx, &aicpuDfxInfo->cpuWaitAicpuNotifyId));
96 0 : CHK_RET(AicpuIndopProcess::AicpuDfxOpInfoInit(aicpuDfxInfo, commId));
97 :
98 0 : return HCCL_SUCCESS;
99 : }
100 :
101 6 : int32_t HcommLocalCopyOnThread(ThreadHandle thread, void *dst, const void *src, uint64_t len)
102 : {
103 6 : PLF_CONFIG_INFO(PLF_DATA_OP, "[%s] thread[0x%llx], dst[0x%llx], src[0x%llx], len[%llu].", __func__, thread, dst, src, len);
104 :
105 6 : CHK_PTR_NULL(dst);
106 5 : CHK_PTR_NULL(src);
107 4 : AddThread(thread);
108 :
109 4 : Thread *const threadPtr = reinterpret_cast<Thread *>(thread);
110 4 : CHK_PTR_NULL(threadPtr);
111 :
112 3 : HcclResult ret = HCCL_SUCCESS;
113 3 : if (threadPtr->IsDeviceA5()) {
114 3 : EXCEPTION_CATCH(ret = threadPtr->LocalCopy(dst, src, len), ret = HCCL_E_INTERNAL);
115 : } else {
116 0 : HcclBuf srcBuf{const_cast<void *>(src), len, nullptr};
117 0 : HcclBuf dstBuf{dst, len, nullptr};
118 0 : Stream *stream = GetStream(thread);
119 0 : CHK_PTR_NULL(stream);
120 0 : ret = HcclLocalCopy(stream, &dstBuf, &srcBuf);
121 : }
122 3 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[%s] FAIL. thread[0x%llx], dst[0x%llx], src[0x%llx], len[%llu].",
123 : __func__, thread, dst, src, len), ret);
124 3 : return HCCL_SUCCESS;
125 : }
126 :
127 6 : int32_t HcommLocalReduceOnThread(ThreadHandle thread, void *dst, const void *src, uint64_t count,
128 : HcommDataType dataType, HcommReduceOp reduceOp)
129 : {
130 6 : PLF_CONFIG_INFO(PLF_DATA_OP, "[%s] thread[0x%llx], dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d].", __func__, thread, dst, src, count, dataType, reduceOp);
131 :
132 6 : CHK_PTR_NULL(dst);
133 5 : CHK_PTR_NULL(src);
134 4 : AddThread(thread);
135 :
136 4 : Thread *const threadPtr = reinterpret_cast<Thread *>(thread);
137 4 : CHK_PTR_NULL(threadPtr);
138 :
139 3 : uint64_t len = count * SIZE_TABLE[dataType];
140 :
141 3 : HcclResult ret = HCCL_SUCCESS;
142 3 : if (threadPtr->IsDeviceA5()) {
143 3 : EXCEPTION_CATCH(ret = threadPtr->LocalReduce(dst, src, len, dataType, reduceOp), ret = HCCL_E_INTERNAL);
144 : } else {
145 0 : CHK_PRT_RET((IsSupportReduce(dataType, reduceOp) == false), HCCL_ERROR("[%s] Not support reduce, "
146 : "dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d]", __func__, dst, src, count, dataType, reduceOp), HCCL_E_PARA);
147 0 : HcclBuf srcBuf{const_cast<void *>(src), len, nullptr};
148 0 : HcclBuf dstBuf{dst, len, nullptr};
149 0 : HcclReduceInfo reduceInfo{static_cast<HcclDataType>(dataType), static_cast<HcclReduceOp>(reduceOp)};
150 0 : Stream *stream = GetStream(thread);
151 0 : CHK_PTR_NULL(stream);
152 0 : ret = HcclLocalCopyReduce(stream, &dstBuf, &srcBuf, reduceInfo);
153 : }
154 3 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[%s] FAIL. thread[0x%llx], dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d].", __func__, thread, dst, src, count, dataType, reduceOp), ret);
155 3 : HCCL_INFO("[%s] SUCCESS.", __func__);
156 3 : return HCCL_SUCCESS;
157 : }
158 :
159 3 : int32_t HcommThreadNotifyRecordOnThread(ThreadHandle thread, ThreadHandle dstThread, uint32_t dstNotifyIdx)
160 : {
161 3 : PLF_CONFIG_INFO(PLF_DATA_OP, "[%s] thread[0x%llx], dstThread[0x%llx], dstNotifyIdx[%u].", __func__, thread, dstThread, dstNotifyIdx);
162 :
163 3 : AddThread(thread);
164 :
165 3 : Thread *const threadPtr = reinterpret_cast<Thread *>(thread);
166 3 : CHK_PTR_NULL(threadPtr);
167 2 : Thread *const dstThreadPtr = reinterpret_cast<Thread *>(dstThread);
168 2 : CHK_PTR_NULL(dstThreadPtr);
169 :
170 1 : HcclResult ret = HCCL_SUCCESS;
171 1 : if (threadPtr->IsDeviceA5()) {
172 1 : LocalNotify *const notifyPtr = dstThreadPtr->GetNotify(dstNotifyIdx);
173 1 : CHK_PTR_NULL(notifyPtr);
174 1 : const uint32_t notifyId = notifyPtr->notifyId_;
175 1 : EXCEPTION_CATCH(ret = threadPtr->LocalNotifyRecord(notifyId), ret = HCCL_E_INTERNAL);
176 : } else {
177 0 : Stream *stream = GetStream(thread);
178 0 : CHK_PTR_NULL(stream);
179 0 : LocalNotify *notify = GetNotify(dstThread, dstNotifyIdx);
180 0 : CHK_PTR_NULL(notify);
181 0 : ret = HcclLocalNotifyRecord(stream, notify);
182 : }
183 1 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[%s] FAIL. thread[0x%llx], dstThread[0x%llx], dstNotifyIdx[%u].", __func__, thread, dstThread, dstNotifyIdx), ret);
184 1 : HCCL_INFO("[%s] SUCCESS.", __func__);
185 1 : return HCCL_SUCCESS;
186 : }
187 :
188 5 : int32_t HcommThreadNotifyWaitOnThread(ThreadHandle thread, uint32_t notifyIdx, uint32_t timeOut)
189 : {
190 5 : PLF_CONFIG_INFO(PLF_DATA_OP, "[%s] thread[0x%llx], notifyIdx[%u], timeOut[%u].", __func__, thread, notifyIdx, timeOut);
191 :
192 5 : AddThread(thread);
193 :
194 5 : Thread *const threadPtr = reinterpret_cast<Thread *>(thread);
195 5 : CHK_PTR_NULL(threadPtr);
196 :
197 4 : HcclResult ret = HCCL_SUCCESS;
198 4 : if (threadPtr->IsDeviceA5()) {
199 4 : LocalNotify *const notifyPtr = threadPtr->GetNotify(notifyIdx);
200 4 : CHK_PTR_NULL(notifyPtr);
201 4 : const uint32_t notifyId = notifyPtr->notifyId_;
202 4 : EXCEPTION_CATCH(ret = threadPtr->LocalNotifyWait(notifyId, timeOut), ret = HCCL_E_INTERNAL);
203 : } else {
204 0 : Stream *stream = GetStream(thread);
205 0 : CHK_PTR_NULL(stream);
206 0 : LocalNotify *notify = GetNotify(thread, notifyIdx);
207 0 : CHK_PTR_NULL(notify);
208 0 : ret = HcclLocalNotifyWait(stream, notify, timeOut);
209 : }
210 4 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[%s] FAIL. thread[0x%llx], notifyIdx[%u], timeOut[%u].", __func__, thread, notifyIdx, timeOut), ret);
211 4 : HCCL_INFO("[%s] SUCCESS.", __func__);
212 4 : return HCCL_SUCCESS;
213 : }
214 :
215 0 : int32_t HcommAclrtNotifyRecordOnThread(ThreadHandle thread, uint64_t dstNotifyId)
216 : {
217 0 : PLF_CONFIG_INFO(PLF_DATA_OP, "[%s] thread[0x%llx], dstNotifyId[%llu].", __func__, thread, dstNotifyId);
218 :
219 0 : AddThread(thread);
220 :
221 0 : Thread *const threadPtr = reinterpret_cast<Thread *>(thread);
222 0 : CHK_PTR_NULL(threadPtr);
223 :
224 0 : HcclResult ret = HCCL_SUCCESS;
225 0 : if (threadPtr->IsDeviceA5()) {
226 0 : EXCEPTION_CATCH(ret = threadPtr->LocalNotifyRecord(dstNotifyId), ret = HCCL_E_INTERNAL);
227 : } else {
228 0 : Stream *stream = GetStream(thread);
229 0 : CHK_PTR_NULL(stream);
230 0 : ret = HcclLocalBareNotifyRecord(stream, dstNotifyId);
231 : }
232 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[%s] FAIL. thread[0x%llx], dstNotifyId[%llu].", __func__, thread, dstNotifyId), ret);
233 0 : HCCL_INFO("[%s] SUCCESS.", __func__);
234 0 : return HCCL_SUCCESS;
235 : }
236 :
237 0 : int32_t HcommAclrtNotifyWaitOnThread(ThreadHandle thread, uint64_t notifyId, uint32_t timeOut)
238 : {
239 0 : PLF_CONFIG_INFO(PLF_DATA_OP, "[%s] thread[0x%llx], notifyId[%llu], timeOut[%u].", __func__, thread, notifyId, timeOut);
240 :
241 0 : AddThread(thread);
242 :
243 0 : Thread *const threadPtr = reinterpret_cast<Thread *>(thread);
244 0 : CHK_PTR_NULL(threadPtr);
245 :
246 0 : HcclResult ret = HCCL_SUCCESS;
247 0 : if (threadPtr->IsDeviceA5()) {
248 0 : EXCEPTION_CATCH(ret = threadPtr->LocalNotifyWait(notifyId, timeOut), ret = HCCL_E_INTERNAL);
249 : } else {
250 0 : Stream *stream = GetStream(thread);
251 0 : CHK_PTR_NULL(stream);
252 0 : ret = HcclLocalBareNotifyWait(stream, notifyId, timeOut);
253 : }
254 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[%s] FAIL. thread[0x%llx], notifyId[%llu], timeOut[%u].", __func__, thread, notifyId, timeOut), ret);
255 0 : HCCL_INFO("[%s] SUCCESS.", __func__);
256 0 : return HCCL_SUCCESS;
257 : }
258 :
259 0 : HcclResult CommTaskPrepare(char *key, uint32_t keyLen) // host ffts+使用
260 : {
261 0 : std::string keyStr = "temp_key";
262 0 : if (key != nullptr && keyLen != 0) {
263 0 : keyStr = std::string(key, keyLen);
264 0 : HCCL_DEBUG("[CommTaskPrepare]key[%s], keyLen[%u]", key, keyLen);
265 : } else {
266 0 : HCCL_DEBUG("[CommTaskPrepare]disable cache, key[0x%llx], keyLen[%u]", key, keyLen);
267 : }
268 :
269 0 : return HcclTaskPrepare(const_cast<char_t*>(keyStr.c_str()), keyStr.length());
270 0 : }
271 :
272 0 : HcclResult CommTaskLaunch(ThreadHandle *threads, uint32_t threadNum) // host ffts+或aicpu stars使用"
273 : {
274 0 : CHK_PTR_NULL(threads);
275 0 : CHK_PRT_RET(threadNum < 1, HCCL_ERROR("[CommTaskLaunch]threadNum is less than 1"), HCCL_E_PARA);
276 :
277 0 : Thread *threadPtr = reinterpret_cast<Thread *>(threads[0]);
278 0 : CHK_PTR_NULL(threadPtr);
279 :
280 0 : if (threadPtr->IsDeviceA5()) {
281 0 : HCCL_INFO("[%s] Running on A5.", __func__);
282 0 : for (uint32_t i = 0; i < threadNum; i++) {
283 0 : Thread *threadPtrLoop = reinterpret_cast<Thread *>(threads[i]);
284 0 : CHK_PTR_NULL(threadPtrLoop);
285 0 : HCCL_INFO("[%s] Launching task in thread[0x%llx].", __func__, threads[i]);
286 0 : EXCEPTION_CATCH(threadPtrLoop->LaunchTask(), return HCCL_E_INTERNAL);
287 : }
288 0 : return HCCL_SUCCESS;
289 : }
290 :
291 0 : std::vector<hccl::Stream> streams;
292 0 : for (uint32_t i = 0; i < threadNum; i++) {
293 0 : hccl::Stream *stream = GetStream(threads[i]);
294 0 : CHK_PTR_NULL(stream);
295 0 : streams.push_back(*stream);
296 : }
297 :
298 0 : return HcclTaskLaunch(streams.data(), threadNum);
299 0 : }
300 :
301 0 : HcclResult DispatchAllStreams(ThreadHandle *threads, uint32_t threadNum)
302 : {
303 0 : CHK_PTR_NULL(threads);
304 0 : CHK_PRT_RET(threadNum < 1, HCCL_ERROR("[DispatchAllStreams]threadNum is less than 1"), HCCL_E_PARA);
305 :
306 0 : Thread *threadPtr = reinterpret_cast<Thread *>(threads[0]);
307 0 : CHK_PTR_NULL(threadPtr);
308 :
309 0 : if (!threadPtr->IsDeviceA5()) {
310 0 : HCCL_ERROR("[%s] DispatchAllStreams is only supported on A5 device.", __func__);
311 0 : return HCCL_E_NOT_SUPPORT;
312 : }
313 :
314 0 : for (uint32_t i = 0; i < threadNum; i++) {
315 0 : Thread *threadPtrLoop = reinterpret_cast<Thread *>(threads[i]);
316 0 : CHK_PTR_NULL(threadPtrLoop);
317 0 : EXCEPTION_CATCH(threadPtrLoop->TryLaunchTask(), return HCCL_E_INTERNAL);
318 : }
319 0 : return HCCL_SUCCESS;
320 : }
321 :
322 : namespace {
323 : // Convert hccl::HcommDataType => Hccl::DataType, hccl::HcommReduceOp => Hccl::ReduceOp
324 :
325 : std::unordered_map<HcommDataType, Hccl::DataType> mapHcommDataTypeToA5 = {
326 : {HcommDataType::HCOMM_DATA_TYPE_INT8, Hccl::DataType::INT8},
327 : {HcommDataType::HCOMM_DATA_TYPE_INT16, Hccl::DataType::INT16},
328 : {HcommDataType::HCOMM_DATA_TYPE_INT32, Hccl::DataType::INT32},
329 : {HcommDataType::HCOMM_DATA_TYPE_FP16, Hccl::DataType::FP16},
330 : {HcommDataType::HCOMM_DATA_TYPE_FP32, Hccl::DataType::FP32},
331 : {HcommDataType::HCOMM_DATA_TYPE_INT64, Hccl::DataType::INT64},
332 : {HcommDataType::HCOMM_DATA_TYPE_UINT64, Hccl::DataType::UINT64},
333 : {HcommDataType::HCOMM_DATA_TYPE_UINT8, Hccl::DataType::UINT8},
334 : {HcommDataType::HCOMM_DATA_TYPE_UINT16, Hccl::DataType::UINT16},
335 : {HcommDataType::HCOMM_DATA_TYPE_UINT32, Hccl::DataType::UINT32},
336 : {HcommDataType::HCOMM_DATA_TYPE_FP64, Hccl::DataType::FP64},
337 : {HcommDataType::HCOMM_DATA_TYPE_BFP16, Hccl::DataType::BFP16},
338 : {HcommDataType::HCOMM_DATA_TYPE_INT128, Hccl::DataType::INT128},
339 : #ifndef OPEN_BUILD_PROJECT
340 : {HcommDataType::HCOMM_DATA_TYPE_HIF8, Hccl::DataType::HIF8},
341 : {HcommDataType::HCOMM_DATA_TYPE_FP8E4M3, Hccl::DataType::FP8E4M3},
342 : {HcommDataType::HCOMM_DATA_TYPE_FP8E5M2, Hccl::DataType::FP8E5M2},
343 : {HcommDataType::HCOMM_DATA_TYPE_FP8E8M0, Hccl::DataType::FP8E8M0}
344 : #endif
345 : };
346 :
347 : std::unordered_map<HcommReduceOp, Hccl::ReduceOp> mapHcommReduceOpToA5 = {
348 : {HcommReduceOp::HCOMM_REDUCE_SUM, Hccl::ReduceOp::SUM},
349 : {HcommReduceOp::HCOMM_REDUCE_PROD, Hccl::ReduceOp::PROD},
350 : {HcommReduceOp::HCOMM_REDUCE_MAX, Hccl::ReduceOp::MAX},
351 : {HcommReduceOp::HCOMM_REDUCE_MIN, Hccl::ReduceOp::MIN}};
352 :
353 7 : inline HcclResult CheckDataTypeAndReduceOp(HcommDataType dataType, HcommReduceOp reduceOp)
354 : {
355 7 : if (mapHcommDataTypeToA5.find(dataType) == mapHcommDataTypeToA5.end()) {
356 3 : HCCL_ERROR("[%s] type[%u] is not supported.", __func__, dataType);
357 3 : return HCCL_E_PARA;
358 : }
359 :
360 4 : if (mapHcommReduceOpToA5.find(reduceOp) == mapHcommReduceOpToA5.end()) {
361 1 : HCCL_ERROR("[%s] op[%u] is not supported.", __func__, reduceOp);
362 1 : return HCCL_E_PARA;
363 : }
364 :
365 3 : return HCCL_SUCCESS;
366 : }
367 :
368 : } // namespace
369 :
370 : // 设置notify wait的等待超时时间，默认单位为秒
371 5 : int32_t HcommSetNotifyWaitTimeOut(float timeOut)
372 : {
373 5 : if (std::isnan(timeOut) || timeOut < 0.0f || timeOut > static_cast<float>(UINT32_MAX)) {
374 0 : HCCL_ERROR("[%s] in aicpu_ts timeOut[%f] is invalid.", __func__, timeOut);
375 0 : return HCCL_E_PARA;
376 : }
377 5 : uint32_t timeOutInt = static_cast<uint32_t>(timeOut);
378 5 : HCCL_INFO("[%s] START in aicpu_ts. timeOut[%u].", __func__, timeOutInt);
379 5 : return g_threadLaunchCtx.SetNotifyWaitTimeOut(timeOutInt);
380 : }
381 :
382 3 : int32_t HcommThreadResAcquireTimeOut(float timeOut)
383 : {
384 3 : if (std::isnan(timeOut) || timeOut < 0.0f || timeOut > static_cast<float>(UINT32_MAX)) {
385 0 : HCCL_ERROR("[%s] in aicpu_ts timeOut[%f] is invalid.", __func__, timeOut);
386 0 : return HCCL_E_PARA;
387 : }
388 3 : uint32_t timeOutInt = static_cast<uint32_t>(timeOut);
389 3 : HCCL_INFO("[%s] START in aicpu_ts. timeOut[%u].", __func__, timeOutInt);
390 3 : return g_threadLaunchCtx.SetSqFullTimeOut(timeOutInt);
391 : }
392 :
393 2 : int32_t HcommChannelNotifyWaitOnThreadWithDefaultTimeout(ThreadHandle thread, ChannelHandle channel, uint32_t localNotifyIdx)
394 : {
395 2 : HCCL_INFO("[%s] START. thread[0x%llx], channel[0x%llx], localNotifyIdx[%u].",
396 : __func__, thread, channel, localNotifyIdx);
397 :
398 : uint32_t notifyWaitTimeOut;
399 2 : g_threadLaunchCtx.GetNotifyWaitTimeOut(notifyWaitTimeOut);
400 :
401 2 : HCCL_DEBUG("[%s] Using default timeout: %u s", __func__, notifyWaitTimeOut);
402 :
403 2 : int32_t ret = HcommChannelNotifyWaitOnThread(thread, channel, localNotifyIdx, notifyWaitTimeOut);
404 2 : if (ret != HCCL_SUCCESS) {
405 0 : HCCL_ERROR("[%s] HcommChannelNotifyWaitOnThread FAILED. thread[0x%llx], channel[0x%llx], localNotifyIdx[%u], ret[%d]",
406 : __func__, thread, channel, localNotifyIdx, ret);
407 0 : return ret;
408 : }
409 :
410 2 : HCCL_INFO("[%s] SUCCESS.", __func__);
411 2 : return HCCL_SUCCESS;
412 : }
413 :
414 3 : int32_t HcommThreadNotifyWaitOnThreadWithDefaultTimeout(ThreadHandle thread, uint32_t notifyIdx)
415 : {
416 3 : HCCL_INFO("[%s] START. thread[0x%llx], notifyIdx[%u].", __func__, thread, notifyIdx);
417 :
418 : uint32_t notifyWaitTimeOut;
419 3 : g_threadLaunchCtx.GetNotifyWaitTimeOut(notifyWaitTimeOut);
420 :
421 3 : HCCL_DEBUG("[%s] Using default timeout: %u s", __func__, notifyWaitTimeOut);
422 :
423 3 : int32_t ret = HcommThreadNotifyWaitOnThread(thread, notifyIdx, notifyWaitTimeOut);
424 3 : if (ret != HCCL_SUCCESS) {
425 0 : HCCL_ERROR("[%s] HcommThreadNotifyWaitOnThread FAILED. thread[0x%llx], notifyIdx[%u], ret[%d]",
426 : __func__, thread, notifyIdx, ret);
427 0 : return ret;
428 : }
429 :
430 3 : HCCL_INFO("[%s] SUCCESS.", __func__);
431 3 : return HCCL_SUCCESS;
432 : }
433 :
434 4 : int32_t HcommWriteOnThread(ThreadHandle thread, ChannelHandle channel, void *dst, const void *src, uint64_t len)
435 : {
436 4 : PLF_CONFIG_INFO(PLF_DATA_OP, "[%s] thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu].",
437 : __func__, thread, channel, dst, src, len);
438 :
439 4 : CHK_PTR_NULL(dst);
440 4 : CHK_PTR_NULL(src);
441 4 : CHK_RET(UnwrapChannelHandle(channel));
442 4 : AddThread(thread);
443 :
444 4 : Thread *const threadPtr = reinterpret_cast<Thread *>(thread);
445 4 : CHK_PTR_NULL(threadPtr);
446 :
447 3 : HcclResult ret = HCCL_SUCCESS;
448 3 : if (threadPtr->IsDeviceA5()) {
449 3 : auto *const transportLitePtr = reinterpret_cast<Hccl::BaseTransportLiteImpl *>(channel);
450 4 : CHK_PTR_NULL(transportLitePtr);
451 3 : auto *const streamLitePtr = static_cast<Hccl::StreamLite *>(threadPtr->GetStreamLitePtr());
452 3 : CHK_PTR_NULL(streamLitePtr);
453 :
454 3 : Hccl::RmaBufferLite locRmaBuf;
455 3 : ret = transportLitePtr->BuildLocRmaBufferLite(reinterpret_cast<uintptr_t>(src), len, locRmaBuf);
456 3 : CHK_PRT_RET(ret != HCCL_SUCCESS,
457 : HCCL_ERROR("[%s] FAIL at BuildLocRmaBufferLite. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu].",
458 : __func__, thread, channel, dst, src, len), ret);
459 2 : const Hccl::Buffer rmtBuf{reinterpret_cast<uintptr_t>(dst), len};
460 :
461 2 : EXCEPTION_CATCH(transportLitePtr->Write(locRmaBuf, rmtBuf, *streamLitePtr), ret = HCCL_E_INTERNAL);
462 2 : } else {
463 0 : HcclBuf locBuf{const_cast<void *>(src), len, nullptr};
464 0 : HcclBuf rmtBuf{dst, len, nullptr};
465 :
466 0 : Stream *stream = GetStream(thread);
467 0 : CHK_PTR_NULL(stream);
468 :
469 0 : ret = HcclRemoteWrite(stream, reinterpret_cast<void *>(channel), &rmtBuf, &locBuf);
470 : }
471 2 : CHK_PRT_RET(ret != HCCL_SUCCESS,
472 : HCCL_ERROR("[%s] FAIL. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu].",
473 : __func__, thread, channel, dst, src, len), ret);
474 1 : HCCL_INFO("[%s] SUCCESS.", __func__);
475 1 : return HCCL_SUCCESS;
476 : }
477 :
478 4 : int32_t HcommWriteReduceOnThread(ThreadHandle thread, ChannelHandle channel, void *dst, const void *src,
479 : uint64_t count, HcommDataType dataType, HcommReduceOp reduceOp)
480 : {
481 4 : PLF_CONFIG_INFO(PLF_DATA_OP, "[%s] thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d].",
482 : __func__, thread, channel, dst, src, count, dataType, reduceOp);
483 :
484 4 : CHK_PTR_NULL(dst);
485 4 : CHK_PTR_NULL(src);
486 4 : CHK_RET(UnwrapChannelHandle(channel));
487 4 : AddThread(thread);
488 :
489 4 : Thread *const threadPtr = reinterpret_cast<Thread *>(thread);
490 4 : CHK_PTR_NULL(threadPtr);
491 :
492 3 : uint64_t len = count * SIZE_TABLE[dataType];
493 :
494 3 : HcclResult ret = HCCL_SUCCESS;
495 3 : if (threadPtr->IsDeviceA5()) {
496 3 : auto *const transportLitePtr = reinterpret_cast<Hccl::BaseTransportLiteImpl *>(channel);
497 5 : CHK_PTR_NULL(transportLitePtr);
498 3 : auto *const streamLitePtr = static_cast<Hccl::StreamLite *>(threadPtr->GetStreamLitePtr());
499 3 : CHK_PTR_NULL(streamLitePtr);
500 :
501 3 : Hccl::RmaBufferLite locRmaBuf;
502 3 : ret = transportLitePtr->BuildLocRmaBufferLite(reinterpret_cast<uintptr_t>(src), len, locRmaBuf);
503 3 : CHK_PRT_RET(ret != HCCL_SUCCESS,
504 : HCCL_ERROR("[%s] FAIL at BuildLocRmaBufferLite. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d].",
505 : __func__, thread, channel, dst, src, count, dataType, reduceOp), ret);
506 2 : const Hccl::Buffer rmtBuf{reinterpret_cast<uintptr_t>(dst), len};
507 :
508 2 : ret = CheckDataTypeAndReduceOp(dataType, reduceOp);
509 2 : CHK_PRT_RET(ret != HCCL_SUCCESS,
510 : HCCL_ERROR("[%s] FAIL at CheckDataTypeAndReduceOp. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d].",
511 : __func__, thread, channel, dst, src, count, dataType, reduceOp), ret);
512 1 : Hccl::ReduceIn reduceIn{mapHcommDataTypeToA5.at(dataType), mapHcommReduceOpToA5.at(reduceOp)};
513 :
514 1 : EXCEPTION_CATCH(transportLitePtr->WriteReduce(locRmaBuf, rmtBuf, reduceIn, *streamLitePtr), ret = HCCL_E_INTERNAL);
515 2 : } else {
516 0 : CHK_PRT_RET((IsSupportReduce(dataType, reduceOp) == false), HCCL_ERROR("[%s] Not support reduce, "
517 : "dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d]", __func__, dst, src, count, dataType, reduceOp), HCCL_E_PARA);
518 0 : HcclBuf locBuf{const_cast<void *>(src), len, nullptr};
519 0 : HcclBuf rmtBuf{dst, len, nullptr};
520 0 : HcclReduceInfo reduceInfo{static_cast<HcclDataType>(dataType), static_cast<HcclReduceOp>(reduceOp)};
521 :
522 0 : Stream *stream = GetStream(thread);
523 0 : CHK_PTR_NULL(stream);
524 :
525 0 : ret = HcclRemoteWriteReduce(stream, reinterpret_cast<void *>(channel), &rmtBuf, &locBuf, reduceInfo);
526 : }
527 1 : CHK_PRT_RET(ret != HCCL_SUCCESS,
528 : HCCL_ERROR("[%s] FAIL. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d].",
529 : __func__, thread, channel, dst, src, count, dataType, reduceOp), ret);
530 1 : HCCL_INFO("[%s] SUCCESS.", __func__);
531 1 : return HCCL_SUCCESS;
532 : }
533 :
534 0 : HcclResult CommWriteReduceWithNotify(ThreadHandle thread, ChannelHandle channel, void *dst, const void *src,
535 : uint64_t count, HcommDataType dataType, HcommReduceOp reduceOp, uint32_t remoteNotifyIdx)
536 : {
537 0 : CHK_PTR_NULL(src);
538 0 : CHK_PTR_NULL(dst);
539 0 : CHK_RET(UnwrapChannelHandle(channel));
540 0 : AddThread(thread);
541 0 : CHK_PRT_RET((IsSupportReduce(dataType, reduceOp) == false), HCCL_ERROR("[%s] Not support reduce, "
542 : "dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d]", __func__, dst, src, count, dataType, reduceOp), HCCL_E_PARA);
543 0 : HcclBuf locBuf{const_cast<void*>(src), count * SIZE_TABLE[dataType], nullptr};
544 0 : HcclBuf rmtBuf{dst, count * SIZE_TABLE[dataType], nullptr};
545 0 : HcclReduceInfo reduceInfo{static_cast<HcclDataType>(dataType), static_cast<HcclReduceOp>(reduceOp)};
546 :
547 0 : Stream *stream = GetStream(thread);
548 0 : CHK_PTR_NULL(stream);
549 :
550 0 : return HcclRemoteWriteReduceWithNotify(stream, reinterpret_cast<void*>(channel), &rmtBuf, &locBuf, reduceInfo,
551 0 : remoteNotifyIdx);
552 : }
553 :
554 6 : int32_t HcommWriteWithNotifyOnThread(ThreadHandle thread, ChannelHandle channel, void *dst, const void *src,
555 : uint64_t len, uint32_t remoteNotifyIdx)
556 : {
557 6 : PLF_CONFIG_INFO(PLF_DATA_OP, "[%s] thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu], remoteNotifyIdx[%u].",
558 : __func__, thread, channel, dst, src, len, remoteNotifyIdx);
559 :
560 6 : CHK_PTR_NULL(dst);
561 5 : CHK_PTR_NULL(src);
562 4 : CHK_RET(UnwrapChannelHandle(channel));
563 3 : AddThread(thread);
564 :
565 3 : Thread *const threadPtr = reinterpret_cast<Thread *>(thread);
566 3 : CHK_PTR_NULL(threadPtr);
567 :
568 2 : HcclResult ret = HCCL_SUCCESS;
569 2 : if (threadPtr->IsDeviceA5()) {
570 2 : HCCL_DEBUG("[%s] Running on A5.", __func__);
571 2 : auto *const transportLitePtr = reinterpret_cast<Hccl::BaseTransportLiteImpl *>(channel);
572 3 : CHK_PTR_NULL(transportLitePtr);
573 2 : auto *const streamLitePtr = static_cast<Hccl::StreamLite *>(threadPtr->GetStreamLitePtr());
574 2 : CHK_PTR_NULL(streamLitePtr);
575 :
576 2 : Hccl::RmaBufferLite locRmaBuf;
577 2 : ret = transportLitePtr->BuildLocRmaBufferLite(reinterpret_cast<uintptr_t>(src), len, locRmaBuf);
578 2 : CHK_PRT_RET(ret != HCCL_SUCCESS,
579 : HCCL_ERROR("[%s] FAIL at BuildLocRmaBufferLite. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu], remoteNotifyIdx[%u].",
580 : __func__, thread, channel, dst, src, len, remoteNotifyIdx), ret);
581 1 : const Hccl::Buffer rmtBuf{reinterpret_cast<uintptr_t>(dst), len};
582 :
583 1 : Hccl::WithNotifyIn withNotify{Hccl::TransportNotifyType::NORMAL, remoteNotifyIdx};
584 :
585 1 : EXCEPTION_CATCH(transportLitePtr->WriteWithNotify(locRmaBuf, rmtBuf, withNotify, *streamLitePtr), ret = HCCL_E_INTERNAL);
586 1 : } else {
587 0 : HcclBuf locBuf{const_cast<void *>(src), len, nullptr};
588 0 : HcclBuf rmtBuf{dst, len, nullptr};
589 :
590 0 : Stream *stream = GetStream(thread);
591 0 : CHK_PTR_NULL(stream);
592 :
593 0 : ret = HcclRemoteWriteWithNotify(stream, reinterpret_cast<void *>(channel), &rmtBuf, &locBuf, remoteNotifyIdx);
594 : }
595 1 : CHK_PRT_RET(ret != HCCL_SUCCESS,
596 : HCCL_ERROR("[%s] FAIL. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu], remoteNotifyIdx[%u].",
597 : __func__, thread, channel, dst, src, len, remoteNotifyIdx), ret);
598 1 : HCCL_INFO("[%s] SUCCESS.", __func__);
599 1 : return HCCL_SUCCESS;
600 : }
601 :
602 8 : int32_t HcommWriteReduceWithNotifyOnThread(ThreadHandle thread, ChannelHandle channel, void *dst,
603 : const void *src, uint64_t count, HcommDataType dataType, HcommReduceOp reduceOp, uint32_t remoteNotifyIdx)
604 : {
605 8 : CHK_PTR_NULL(dst);
606 7 : CHK_PTR_NULL(src);
607 6 : CHK_RET(UnwrapChannelHandle(channel));
608 :
609 5 : PLF_CONFIG_INFO(PLF_DATA_OP, "[%s] thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d], remoteNotifyIdx[%u].",
610 : __func__, thread, channel, dst, src, count, dataType, reduceOp, remoteNotifyIdx);
611 :
612 5 : AddThread(thread);
613 :
614 5 : Thread *const threadPtr = reinterpret_cast<Thread *>(thread);
615 5 : CHK_PTR_NULL(threadPtr);
616 :
617 4 : uint64_t len = count * SIZE_TABLE[dataType];
618 :
619 4 : HcclResult ret = HCCL_SUCCESS;
620 4 : if (threadPtr->IsDeviceA5()) {
621 4 : HCCL_DEBUG("[%s] Running on A5.", __func__);
622 4 : auto *const transportLitePtr = reinterpret_cast<Hccl::BaseTransportLiteImpl *>(channel);
623 7 : CHK_PTR_NULL(transportLitePtr);
624 4 : auto *const streamLitePtr = static_cast<Hccl::StreamLite *>(threadPtr->GetStreamLitePtr());
625 4 : CHK_PTR_NULL(streamLitePtr);
626 :
627 4 : Hccl::RmaBufferLite locRmaBuf;
628 4 : ret = transportLitePtr->BuildLocRmaBufferLite(reinterpret_cast<uintptr_t>(src), len, locRmaBuf);
629 4 : CHK_PRT_RET(ret != HCCL_SUCCESS,
630 : HCCL_ERROR("[%s] FAIL at BuildLocRmaBufferLite. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d], remoteNotifyIdx[%u].",
631 : __func__, thread, channel, dst, src, count, dataType, reduceOp, remoteNotifyIdx), ret);
632 3 : const Hccl::Buffer rmtBuf{reinterpret_cast<uintptr_t>(dst), len};
633 :
634 3 : ret = CheckDataTypeAndReduceOp(dataType, reduceOp);
635 3 : CHK_PRT_RET(ret != HCCL_SUCCESS,
636 : HCCL_ERROR("[%s] FAIL at CheckDataTypeAndReduceOp. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d], remoteNotifyIdx[%u].",
637 : __func__, thread, channel, dst, src, count, dataType, reduceOp, remoteNotifyIdx), ret);
638 1 : Hccl::ReduceIn reduceIn{mapHcommDataTypeToA5.at(dataType), mapHcommReduceOpToA5.at(reduceOp)};
639 :
640 1 : Hccl::WithNotifyIn withNotify{Hccl::TransportNotifyType::NORMAL, remoteNotifyIdx};
641 :
642 1 : EXCEPTION_CATCH(transportLitePtr->WriteReduceWithNotify(locRmaBuf, rmtBuf, reduceIn, withNotify, *streamLitePtr), ret = HCCL_E_INTERNAL);
643 3 : } else {
644 0 : ret = HCCL_E_NOT_SUPPORT;
645 : }
646 1 : CHK_PRT_RET(ret != HCCL_SUCCESS,
647 : HCCL_ERROR("[%s] FAIL. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d], remoteNotifyIdx[%u].",
648 : __func__, thread, channel, dst, src, count, dataType, reduceOp, remoteNotifyIdx), ret);
649 1 : HCCL_INFO("[%s] SUCCESS.", __func__);
650 1 : return HCCL_SUCCESS;
651 : }
652 :
653 3 : int32_t HcommReadOnThread(ThreadHandle thread, ChannelHandle channel, void *dst, const void *src, uint64_t len)
654 : {
655 3 : CHK_PTR_NULL(dst);
656 3 : CHK_PTR_NULL(src);
657 3 : CHK_RET(UnwrapChannelHandle(channel));
658 :
659 3 : PLF_CONFIG_INFO(PLF_DATA_OP, "[%s] thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu].",
660 : __func__, thread, channel, dst, src, len);
661 :
662 3 : AddThread(thread);
663 :
664 3 : Thread *const threadPtr = reinterpret_cast<Thread *>(thread);
665 3 : CHK_PTR_NULL(threadPtr);
666 :
667 2 : HcclResult ret = HCCL_SUCCESS;
668 2 : if (threadPtr->IsDeviceA5()) {
669 2 : auto *const transportLitePtr = reinterpret_cast<Hccl::BaseTransportLiteImpl *>(channel);
670 3 : CHK_PTR_NULL(transportLitePtr);
671 2 : auto *const streamLitePtr = static_cast<Hccl::StreamLite *>(threadPtr->GetStreamLitePtr());
672 2 : CHK_PTR_NULL(streamLitePtr);
673 :
674 2 : Hccl::RmaBufferLite locRmaBuf;
675 2 : ret = transportLitePtr->BuildLocRmaBufferLite(reinterpret_cast<uintptr_t>(dst), len, locRmaBuf);
676 2 : CHK_PRT_RET(ret != HCCL_SUCCESS,
677 : HCCL_ERROR("[%s] FAIL at BuildLocRmaBufferLite. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu].",
678 : __func__, thread, channel, dst, src, len), ret);
679 1 : const Hccl::Buffer rmtBuf{reinterpret_cast<uintptr_t>(src), len};
680 :
681 1 : EXCEPTION_CATCH(transportLitePtr->Read(locRmaBuf, rmtBuf, *streamLitePtr), ret = HCCL_E_INTERNAL);
682 1 : } else {
683 0 : HcclBuf locBuf{dst, len, nullptr};
684 0 : HcclBuf rmtBuf{const_cast<void *>(src), len, nullptr};
685 :
686 0 : Stream *stream = GetStream(thread);
687 0 : CHK_PTR_NULL(stream);
688 :
689 0 : ret = HcclRemoteRead(stream, reinterpret_cast<void *>(channel), &locBuf, &rmtBuf);
690 : }
691 1 : CHK_PRT_RET(ret != HCCL_SUCCESS,
692 : HCCL_ERROR("[%s] FAIL. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu].",
693 : __func__, thread, channel, dst, src, len), ret);
694 1 : HCCL_INFO("[%s] SUCCESS.", __func__);
695 1 : return HCCL_SUCCESS;
696 : }
697 :
698 4 : int32_t HcommReadReduceOnThread(ThreadHandle thread, ChannelHandle channel, void *dst, const void *src,
699 : uint64_t count, HcommDataType dataType, HcommReduceOp reduceOp)
700 : {
701 4 : CHK_PTR_NULL(dst);
702 4 : CHK_PTR_NULL(src);
703 4 : CHK_RET(UnwrapChannelHandle(channel));
704 :
705 4 : PLF_CONFIG_INFO(PLF_DATA_OP, "[%s] thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d].",
706 : __func__, thread, channel, dst, src, count, dataType, reduceOp);
707 :
708 4 : AddThread(thread);
709 :
710 4 : Thread *const threadPtr = reinterpret_cast<Thread *>(thread);
711 4 : CHK_PTR_NULL(threadPtr);
712 :
713 3 : uint64_t len = count * SIZE_TABLE[dataType];
714 :
715 3 : HcclResult ret = HCCL_SUCCESS;
716 3 : if (threadPtr->IsDeviceA5()) {
717 3 : auto *const transportLitePtr = reinterpret_cast<Hccl::BaseTransportLiteImpl *>(channel);
718 5 : CHK_PTR_NULL(transportLitePtr);
719 3 : auto *const streamLitePtr = static_cast<Hccl::StreamLite *>(threadPtr->GetStreamLitePtr());
720 3 : CHK_PTR_NULL(streamLitePtr);
721 :
722 3 : Hccl::RmaBufferLite locRmaBuf;
723 3 : ret = transportLitePtr->BuildLocRmaBufferLite(reinterpret_cast<uintptr_t>(dst), len, locRmaBuf);
724 3 : CHK_PRT_RET(ret != HCCL_SUCCESS,
725 : HCCL_ERROR("[%s] FAIL at BuildLocRmaBufferLite. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d].",
726 : __func__, thread, channel, dst, src, count, dataType, reduceOp), ret);
727 2 : const Hccl::Buffer rmtBuf{reinterpret_cast<uintptr_t>(src), len};
728 :
729 2 : ret = CheckDataTypeAndReduceOp(dataType, reduceOp);
730 2 : CHK_PRT_RET(ret != HCCL_SUCCESS,
731 : HCCL_ERROR("[%s] FAIL at CheckDataTypeAndReduceOp. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d].",
732 : __func__, thread, channel, dst, src, count, dataType, reduceOp), ret);
733 1 : Hccl::ReduceIn reduceIn{mapHcommDataTypeToA5.at(dataType), mapHcommReduceOpToA5.at(reduceOp)};
734 :
735 1 : EXCEPTION_CATCH(transportLitePtr->ReadReduce(locRmaBuf, rmtBuf, reduceIn, *streamLitePtr), ret = HCCL_E_INTERNAL);
736 2 : } else {
737 0 : CHK_PRT_RET((IsSupportReduce(dataType, reduceOp) == false), HCCL_ERROR("[%s] Not support reduce, "
738 : "dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d]", __func__, dst, src, count, dataType, reduceOp), HCCL_E_PARA);
739 0 : HcclBuf locBuf{dst, len, nullptr};
740 0 : HcclBuf rmtBuf{const_cast<void *>(src), len, nullptr};
741 0 : HcclReduceInfo reduceInfo{static_cast<HcclDataType>(dataType), static_cast<HcclReduceOp>(reduceOp)};
742 :
743 0 : Stream *stream = GetStream(thread);
744 0 : CHK_PTR_NULL(stream);
745 :
746 0 : ret = HcclRemoteReadReduce(stream, reinterpret_cast<void *>(channel), &locBuf, &rmtBuf, reduceInfo);
747 : }
748 1 : CHK_PRT_RET(ret != HCCL_SUCCESS,
749 : HCCL_ERROR("[%s] FAIL. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d].",
750 : __func__, thread, channel, dst, src, count, dataType, reduceOp), ret);
751 1 : HCCL_INFO("[%s] SUCCESS.", __func__);
752 1 : return HCCL_SUCCESS;
753 : }
754 :
755 16 : int32_t HcommBatchTransferOnThread(ThreadHandle thread, ChannelHandle channel,
756 : const HcommBatchTransferDesc *transferDescs, uint32_t transferDescNum)
757 : {
758 16 : CHK_PTR_NULL(transferDescs);
759 15 : CHK_RET(UnwrapChannelHandle(channel));
760 :
761 14 : PLF_CONFIG_INFO(PLF_DATA_OP, "[%s] thread[0x%llx], channel[0x%llx], transferDescNum[%u].",
762 : __func__, thread, channel, transferDescNum);
763 :
764 14 : CHK_PRT_RET(transferDescNum == 0,
765 : HCCL_ERROR("[%s] transferDescNum is 0.", __func__), HCCL_E_PARA);
766 :
767 13 : Thread *const threadPtr = reinterpret_cast<Thread *>(thread);
768 13 : CHK_PTR_NULL(threadPtr);
769 12 : AddThread(thread);
770 12 : HcclResult ret = HCCL_SUCCESS;
771 12 : if (threadPtr->IsDeviceA5()) {
772 12 : auto *const ubTransportLitePtr = reinterpret_cast<Hccl::UbTransportLiteImpl *>(channel);
773 12 : CHK_PTR_NULL(ubTransportLitePtr);
774 12 : auto *const streamLitePtr = static_cast<Hccl::StreamLite *>(threadPtr->GetStreamLitePtr());
775 12 : CHK_PTR_NULL(streamLitePtr);
776 12 : ret = ubTransportLitePtr->ExecuteBatchTransfer(streamLitePtr, transferDescs, transferDescNum);
777 : } else {
778 0 : Stream *stream = GetStream(thread);
779 0 : CHK_PTR_NULL(stream);
780 0 : hccl::Transport *transport = reinterpret_cast<hccl::Transport *>(channel);
781 0 : CHK_PTR_NULL(transport);
782 0 : ret = transport->BatchTransferAsync(transferDescs, transferDescNum, *stream);
783 0 : if (ret == HCCL_E_NOT_SUPPORT) {
784 0 : return HCCL_E_NOT_SUPPORT;
785 : }
786 : }
787 12 : CHK_PRT_RET(ret != HCCL_SUCCESS,
788 : HCCL_ERROR("[%s] BatchTransferAsync failed.", __func__), ret);
789 :
790 8 : HCCL_INFO("[%s] SUCCESS. transferDescNum[%u].", __func__, transferDescNum);
791 8 : return ret;
792 : }
793 :
794 0 : int32_t HcommWriteNbiOnThread(ThreadHandle thread, ChannelHandle channel, void *dst, const void *src, uint64_t len)
795 : {
796 0 : HCCL_DEBUG("[%s] thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu].", __func__, thread, channel, dst, src, len);
797 0 : CHK_PTR_NULL(src);
798 0 : CHK_PTR_NULL(dst);
799 0 : return HCCL_E_NOT_SUPPORT;
800 : }
801 :
802 0 : int32_t HcommWriteNbi(ChannelHandle channel, void *dst, const void *src, uint64_t len)
803 : {
804 0 : HCCL_DEBUG("[%s] channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu].", __func__, channel, dst, src, len);
805 0 : CHK_PTR_NULL(src);
806 0 : CHK_PTR_NULL(dst);
807 0 : return HCCL_E_NOT_SUPPORT;
808 : }
809 :
810 0 : int32_t HcommWriteWithNotifyNbiOnThread(ThreadHandle thread, ChannelHandle channel, void *dst, const void *src,
811 : uint64_t len, uint32_t remoteNotifyIdx)
812 : {
813 0 : HCCL_DEBUG("[%s] thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu], remoteNotifyIdx[%u].",
814 : __func__, thread, channel, dst, src, len, remoteNotifyIdx);
815 0 : CHK_PTR_NULL(src);
816 0 : CHK_PTR_NULL(dst);
817 0 : return HCCL_E_NOT_SUPPORT;
818 : }
819 :
820 0 : int32_t HcommWriteWithNotifyNbi(ChannelHandle channel, void *dst, const void *src,
821 : uint64_t len, uint32_t remoteNotifyIdx)
822 : {
823 0 : HCCL_DEBUG("[%s] channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu], remoteNotifyIdx[%u].",
824 : __func__, channel, dst, src, len, remoteNotifyIdx);
825 0 : CHK_PTR_NULL(src);
826 0 : CHK_PTR_NULL(dst);
827 0 : return HCCL_E_NOT_SUPPORT;
828 : }
829 :
830 0 : int32_t HcommReadNbiOnThread(ThreadHandle thread, ChannelHandle channel, void *dst, const void *src, uint64_t len)
831 : {
832 0 : HCCL_DEBUG("[%s] thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu].", __func__, thread, channel, dst, src, len);
833 0 : CHK_PTR_NULL(src);
834 0 : CHK_PTR_NULL(dst);
835 0 : return HCCL_E_NOT_SUPPORT;
836 : }
837 :
838 0 : int32_t HcommReadNbi(ChannelHandle channel, void *dst, const void *src, uint64_t len)
839 : {
840 0 : HCCL_DEBUG("[%s] channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu].", __func__, channel, dst, src, len);
841 0 : CHK_PTR_NULL(src);
842 0 : CHK_PTR_NULL(dst);
843 0 : return HCCL_E_NOT_SUPPORT;
844 : }
845 :
846 3 : int32_t HcommChannelNotifyRecordOnThread(ThreadHandle thread, ChannelHandle channel, uint32_t remoteNotifyIdx)
847 : {
848 3 : CHK_RET(UnwrapChannelHandle(channel));
849 :
850 2 : PLF_CONFIG_INFO(PLF_DATA_OP, "[%s] thread[0x%llx], channel[0x%llx], remoteNotifyIdx[%u].", __func__, thread, channel, remoteNotifyIdx);
851 :
852 2 : AddThread(thread);
853 :
854 2 : Thread *const threadPtr = reinterpret_cast<Thread *>(thread);
855 2 : CHK_PTR_NULL(threadPtr);
856 :
857 1 : HcclResult ret = HCCL_SUCCESS;
858 1 : if (threadPtr->IsDeviceA5()) {
859 1 : HCCL_DEBUG("[%s] Running on A5.", __func__);
860 1 : auto *const transportLitePtr = reinterpret_cast<Hccl::BaseTransportLiteImpl *>(channel);
861 1 : CHK_PTR_NULL(transportLitePtr);
862 1 : auto *const streamLitePtr = static_cast<Hccl::StreamLite *>(threadPtr->GetStreamLitePtr());
863 1 : CHK_PTR_NULL(streamLitePtr);
864 1 : HCCL_INFO("channel streamlite ptr %p.", streamLitePtr);
865 :
866 1 : EXCEPTION_CATCH(transportLitePtr->Post(remoteNotifyIdx, *streamLitePtr), ret = HCCL_E_INTERNAL);
867 : } else {
868 0 : Stream *stream = GetStream(thread);
869 0 : CHK_PTR_NULL(stream);
870 :
871 0 : ret = HcclRemoteNotifyRecord(stream, reinterpret_cast<void *>(channel), remoteNotifyIdx);
872 : }
873 1 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[%s] FAIL. thread[0x%llx], channel[0x%llx], remoteNotifyIdx[%u].", __func__, thread, channel, remoteNotifyIdx), ret);
874 1 : HCCL_INFO("[%s] SUCCESS.", __func__);
875 1 : return HCCL_SUCCESS;
876 : }
877 :
878 0 : int32_t HcommChannelNotifyRecord(ChannelHandle channel, uint32_t remoteNotifyIdx)
879 : {
880 0 : HCCL_DEBUG("[%s] channel[0x%llx], remoteNotifyIdx[%u].", __func__, channel, remoteNotifyIdx);
881 0 : return HCCL_E_NOT_SUPPORT;
882 : }
883 :
884 10 : int32_t HcommChannelNotifyWaitOnThread(ThreadHandle thread, ChannelHandle channel, uint32_t localNotifyIdx, uint32_t timeOut)
885 : {
886 10 : CHK_RET(UnwrapChannelHandle(channel));
887 :
888 8 : PLF_CONFIG_INFO(PLF_DATA_OP, "[%s] thread[0x%llx], channel[0x%llx], localNotifyIdx[%u], timeOut[%u].", __func__, thread, channel, localNotifyIdx, timeOut);
889 :
890 8 : AddThread(thread);
891 :
892 8 : Thread *const threadPtr = reinterpret_cast<Thread *>(thread);
893 8 : CHK_PTR_NULL(threadPtr);
894 :
895 6 : HcclResult ret = HCCL_SUCCESS;
896 6 : if (threadPtr->IsDeviceA5()) {
897 6 : HCCL_DEBUG("[%s] Running on A5.", __func__);
898 6 : auto *const transportLitePtr = reinterpret_cast<Hccl::BaseTransportLiteImpl *>(channel);
899 6 : CHK_PTR_NULL(transportLitePtr);
900 6 : auto *const streamLitePtr = static_cast<Hccl::StreamLite *>(threadPtr->GetStreamLitePtr());
901 6 : CHK_PTR_NULL(streamLitePtr);
902 :
903 6 : EXCEPTION_CATCH(transportLitePtr->WaitWithTimeout(localNotifyIdx, *streamLitePtr, timeOut), ret = HCCL_E_INTERNAL);
904 : } else {
905 0 : Stream *stream = GetStream(thread);
906 0 : CHK_PTR_NULL(stream);
907 :
908 0 : ret = HcclRemoteNotifyWait(stream, reinterpret_cast<void *>(channel), localNotifyIdx, timeOut);
909 : }
910 6 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[%s] FAIL. thread[0x%llx], channel[0x%llx], localNotifyIdx[%u], timeOut[%u].", __func__, thread, channel, localNotifyIdx, timeOut), ret);
911 6 : HCCL_INFO("[%s] SUCCESS.", __func__);
912 6 : return HCCL_SUCCESS;
913 : }
914 :
915 0 : int32_t HcommChannelNotifyWait(ChannelHandle channel, uint32_t localNotifyIdx, uint32_t timeOut)
916 : {
917 0 : HCCL_DEBUG("[%s] channel[0x%llx], localNotifyIdx[%u], timeOut[%u].", __func__, channel, localNotifyIdx, timeOut);
918 0 : return HCCL_E_NOT_SUPPORT;
919 : }
920 :
921 0 : HcclResult CommFence(ThreadHandle thread, ChannelHandle channel) // 控制前后的任务保序
922 : {
923 0 : CHK_RET(UnwrapChannelHandle(channel));
924 :
925 0 : HCCL_DEBUG("[CommFence] thread[0x%llx], channel[0x%llx].", thread, channel);
926 0 : Stream *stream = GetStream(thread);
927 0 : CHK_PTR_NULL(stream);
928 :
929 0 : return HcclRemoteFence(stream, reinterpret_cast<void *>(channel), false);
930 : }
931 :
932 0 : int32_t HcommSetLaunchMode(const char *launchTag, HcommLaunchMode mode)
933 : {
934 0 : HCCL_DEBUG("HcommSetLaunchMode launchTag[%s]", launchTag);
935 0 : return g_threadLaunchCtx.SetLaunchMode(launchTag, mode);
936 : }
937 :
938 0 : int32_t HcommBatchModeStart(const char *batchTag)
939 : {
940 0 : return HcommSetLaunchMode(batchTag, HCOMM_LAUNCH_MODE_BATCH);
941 : }
942 :
943 0 : int32_t HcommBatchModeEnd(const char *batchTag)
944 : {
945 0 : return HcommSetLaunchMode(batchTag, HCOMM_LAUNCH_MODE_EAGER);
946 : }
947 :
948 0 : int32_t HcommAcquireComm(const char* commId)
949 : {
950 0 : CHK_PTR_NULL(commId);
951 : DevType deviceType;
952 0 : CHK_RET(hrtGetDeviceType(deviceType));
953 0 : HCCL_INFO("[%s]comId[%s], devType[%d]", __func__, commId, deviceType);
954 0 : if (deviceType != DevType::DEV_TYPE_950 && deviceType != DevType::DEV_TYPE_960) {
955 0 : HcclCommAicpu *hcclComm = AicpuHcclProcess::AicpuGetCommbyGroup(commId);
956 0 : CHK_PRT_RET(!hcclComm, HCCL_ERROR("%s AicpuGetCommbyGroup is null, commId[%s]", __func__, commId), HCCL_E_PTR);
957 0 : CHK_RET(hcclComm->SetDispatcherCtxOnThread());
958 0 : } else {
959 0 : CollCommAicpuMgr *hcclComm = AicpuIndopProcess::AicpuGetCommMgrbyGroup(commId);
960 0 : CHK_PRT_RET(!hcclComm, HCCL_ERROR("%s AicpuGetCommMgrbyGroup is null, commId[%s]", __func__, commId), HCCL_E_PTR);
961 : }
962 0 : return HCCL_SUCCESS;
963 : }
964 :
965 0 : int32_t HcommChannelRegisterDfx(ChannelHandle channel, std::function<HcclResult(u32, u32, const Hccl::TaskParam&, u64)> callback) {
966 0 : CHK_RET(UnwrapChannelHandle(channel));
967 :
968 0 : auto *const transportLitePtr = reinterpret_cast<Hccl::BaseTransportLiteImpl *>(channel);
969 0 : CHK_PTR_NULL(transportLitePtr);
970 0 : CHK_RET(transportLitePtr->SetAddTaskInfoCallback(callback));
971 0 : HCCL_INFO("[HcommChannelRegisterDfx] ChannelHandle[0x%llx] Init success", channel);
972 0 : return HCCL_SUCCESS;
973 : }
974 :
975 0 : int32_t HcommThreadRegisterDfx(ThreadHandle thread, std::function<HcclResult(u32, u32, const Hccl::TaskParam&, u64)> callback) {
976 0 : Thread *threadPtr = reinterpret_cast<Thread *>(thread);
977 0 : CHK_PTR_NULL(threadPtr);
978 0 : CHK_RET(threadPtr->SetAddTaskInfoCallback(callback));
979 0 : HCCL_INFO("[HcommThreadRegisterDfx] ThreadHandle[0x%llx] Init success", thread);
980 0 : return HCCL_SUCCESS;
981 : }
982 :
983 0 : int32_t HcommThreadRegisterCheckExecStatus(ThreadHandle thread, std::function<HcclResult(bool)> callback)
984 : {
985 0 : Thread *threadPtr = reinterpret_cast<Thread *>(thread);
986 0 : CHK_PTR_NULL(threadPtr);
987 0 : CHK_RET(threadPtr->SetCheckExecStatusCallback(callback));
988 0 : return HCCL_SUCCESS;
989 : }
990 :
991 0 : int32_t HcommReleaseComm(const char* commId)
992 : {
993 0 : CHK_PTR_NULL(commId);
994 : DevType deviceType;
995 0 : CHK_RET(hrtGetDeviceType(deviceType));
996 0 : HCCL_INFO("[%s]comId[%s], devType[%d]", __func__, commId, deviceType);
997 0 : if (deviceType != DevType::DEV_TYPE_950 && deviceType != DevType::DEV_TYPE_960) {
998 0 : AicpuHcclProcess::AicpuReleaseCommbyGroup(commId);
999 : } else {
1000 0 : AicpuIndopProcess::AicpuReleaseCommMgrbyGroup(commId);
1001 : }
1002 0 : return HCCL_SUCCESS;
1003 : }
1004 :
1005 0 : int32_t HcommFenceOnThread(ThreadHandle thread)
1006 : {
1007 0 : HCCL_DEBUG("[%s] thread[0x%llx].", __func__, thread);
1008 0 : return HCCL_E_NOT_SUPPORT;
1009 : }
1010 :
1011 : #ifdef __cplusplus
1012 : extern "C" {
1013 : #endif // __cplusplus
1014 0 : int32_t HcommFlush()
1015 : {
1016 0 : return HCCL_E_NOT_SUPPORT;
1017 : }
1018 :
1019 3 : int32_t HcommChannelFenceOnThread(ThreadHandle thread, ChannelHandle channel)
1020 : {
1021 3 : CHK_RET(UnwrapChannelHandle(channel));
1022 :
1023 2 : HCCL_DEBUG("[%s] thread[0x%llx], channel[0x%llx].", __func__, thread, channel);
1024 2 : Thread *const threadPtr = reinterpret_cast<Thread *>(thread);
1025 2 : CHK_PTR_NULL(threadPtr);
1026 1 : if (threadPtr->IsDeviceA5()) {
1027 1 : auto *const transportLitePtr = reinterpret_cast<Hccl::BaseTransportLiteImpl *>(channel);
1028 1 : CHK_PTR_NULL(transportLitePtr);
1029 1 : CHK_RET(transportLitePtr->Fence());
1030 : }
1031 :
1032 1 : return HCCL_SUCCESS;
1033 : }
1034 :
1035 0 : int32_t HcommChannelFence(ChannelHandle channel)
1036 : {
1037 0 : HCCL_DEBUG("[%s] channel[0x%llx].", __func__, channel);
1038 0 : return HCCL_E_NOT_SUPPORT;
1039 : }
1040 :
1041 0 : int32_t HcommThreadJoin(ThreadHandle thread, uint32_t timeout)
1042 : {
1043 0 : hccl::Thread *threadPtr = reinterpret_cast<hccl::Thread *>(thread);
1044 0 : CHK_PTR_NULL(threadPtr);
1045 :
1046 0 : HCCL_INFO("[%s] START. thread[0x%llx].", __func__, thread);
1047 :
1048 0 : if (threadPtr->IsDeviceA5()) {
1049 0 : HCCL_INFO("[%s] Running on A5.", __func__);
1050 0 : auto *const streamLitePtr = static_cast<Hccl::StreamLite *>(threadPtr->GetStreamLitePtr());
1051 0 : CHK_PTR_NULL(streamLitePtr);
1052 0 : auto *const rtsqPtr = streamLitePtr->GetRtsq();
1053 0 : CHK_PTR_NULL(rtsqPtr);
1054 :
1055 0 : uint32_t head = 0;
1056 0 : uint32_t tail = 0;
1057 0 : uint32_t sqId = streamLitePtr->GetSqId();
1058 0 : EXCEPTION_CATCH(tail = rtsqPtr->QuerySqTail(), return HCCL_E_INTERNAL);
1059 0 : HCCL_INFO("[%s] aicpu stream sqid[%u] tail[%u]", __func__, sqId, tail);
1060 :
1061 0 : u64 startUsec = GetCurAicpuTimestamp();
1062 0 : u64 lastUsec = startUsec;
1063 0 : constexpr uint64_t NANOSECOND_TO_SECOND = 1000000000U;
1064 0 : const uint64_t kPrintSqInterval = 30U;
1065 : do {
1066 0 : EXCEPTION_CATCH(head = rtsqPtr->QuerySqHead(), return HCCL_E_INTERNAL);
1067 0 : u64 curUsec = GetCurAicpuTimestamp();
1068 0 : if (curUsec - startUsec > NANOSECOND_TO_SECOND * timeout) {
1069 0 : HCCL_ERROR("[%s] timeout %us. curhead:%u, curtail:%u, sqId:%u",
1070 : __func__, timeout, head, tail, sqId);
1071 0 : return HCCL_E_TIMEOUT;
1072 : }
1073 :
1074 : // 等待下发阶段，每隔30s打印一次状态
1075 0 : if (curUsec - lastUsec > NANOSECOND_TO_SECOND * kPrintSqInterval) {
1076 0 : lastUsec = curUsec;
1077 0 : HCCL_RUN_INFO("[%s]Current state. sqid:%d, head:%u, tail:%u",
1078 : __func__, sqId, head, tail);
1079 : }
1080 0 : } while (head != tail);
1081 0 : HCCL_INFO("[%s] SUCCESS. RTSQ's head[%u] == tail[%u].", __func__, head, tail);
1082 0 : return HCCL_SUCCESS;
1083 : }
1084 :
1085 0 : HCCL_ERROR("[%s]Does not support this interface.", __func__);
1086 0 : return HCCL_E_NOT_SUPPORT;
1087 : }
1088 :
1089 0 : int32_t HcommChannelDrainOnThread(ThreadHandle thread, ChannelHandle channel)
1090 : {
1091 0 : CHK_RET(UnwrapChannelHandle(channel));
1092 :
1093 0 : AddThread(thread);
1094 0 : Thread *const threadPtr = reinterpret_cast<Thread *>(thread);
1095 0 : CHK_PTR_NULL(threadPtr);
1096 0 : Stream *stream = GetStream(thread);
1097 0 : CHK_PTR_NULL(stream);
1098 :
1099 0 : HcclResult ret = HCCL_SUCCESS;
1100 0 : if (threadPtr->IsDeviceA5()) {
1101 0 : HCCL_DEBUG("[%s] Running on A5.", __func__);
1102 0 : auto *const transportLitePtr = reinterpret_cast<Hccl::BaseTransportLiteImpl *>(channel);
1103 0 : CHK_PTR_NULL(transportLitePtr);
1104 0 : auto *const streamLitePtr = static_cast<Hccl::StreamLite *>(threadPtr->GetStreamLitePtr());
1105 0 : CHK_PTR_NULL(streamLitePtr);
1106 :
1107 0 : EXCEPTION_CATCH(transportLitePtr->Drain(*streamLitePtr), ret = HCCL_E_INTERNAL);
1108 0 : return ret;
1109 : }
1110 :
1111 0 : ret = HcclRemoteDrain(stream, reinterpret_cast<void *>(channel));
1112 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
1113 : HCCL_ERROR("[%s] Run FAIL. thread[0x%llx], channel[0x%llx].", __func__, thread, channel), ret);
1114 :
1115 0 : return HCCL_SUCCESS;
1116 : }
1117 : #ifdef __cplusplus
1118 : }
1119 : #endif // __cplusplus
1120 :
1121 0 : HcclResult HcommProfilingReportDeviceOp(const char* groupname) {
1122 0 : if (!GetProfilingEnable()) {
1123 0 : return HCCL_SUCCESS;
1124 : }
1125 0 : CHK_PTR_NULL(groupname);
1126 :
1127 : DevType deviceType;
1128 0 : CHK_RET(hrtGetDeviceType(deviceType));
1129 0 : if (deviceType != DevType::DEV_TYPE_950 && deviceType != DevType::DEV_TYPE_960) {
1130 0 : return HCCL_SUCCESS;
1131 : }
1132 :
1133 0 : CHK_RET(AicpuIndopProcess::ProfilingReportDeviceOp());
1134 0 : return HCCL_SUCCESS;
1135 : }
1136 :
1137 0 : HcclResult HcommProfilingReportKernelStartTask(uint64_t thread, const char* groupname)
1138 : {
1139 0 : if (!GetProfilingEnable()) {
1140 0 : return HCCL_SUCCESS;
1141 : }
1142 :
1143 : DevType deviceType;
1144 0 : CHK_RET(hrtGetDeviceType(deviceType));
1145 0 : if (deviceType != DevType::DEV_TYPE_950 && deviceType != DevType::DEV_TYPE_960) {
1146 0 : return HCCL_SUCCESS;
1147 : }
1148 0 : CHK_PTR_NULL(groupname);
1149 0 : CHK_RET(AicpuIndopProcess::UpdateTask(groupname));
1150 0 : Thread *const threadPtr = reinterpret_cast<Thread *>(thread);
1151 0 : CHK_PTR_NULL(threadPtr);
1152 0 : auto *const streamLitePtr = static_cast<Hccl::StreamLite *>(threadPtr->GetStreamLitePtr());
1153 0 : CHK_PTR_NULL(streamLitePtr);
1154 0 : Hccl::FlagTaskInfo flagTaskInfo;
1155 0 : flagTaskInfo.taskId = streamLitePtr->GetRtsq()->GetTaskId();
1156 0 : flagTaskInfo.type = Hccl::MainStreamTaskType::HEAD;
1157 0 : Hccl::ProfilingHandlerLite::GetInstance().ReportMainStreamTask(flagTaskInfo);
1158 0 : HCCL_INFO("[%s] END, thread [%llu], groupname[%s], taskId[%u].", __func__, thread, groupname, flagTaskInfo.taskId);
1159 0 : return HCCL_SUCCESS;
1160 : }
1161 :
1162 0 : HcclResult HcommProfilingReportKernelEndTask(uint64_t thread, const char* groupname)
1163 : {
1164 0 : if (!GetProfilingEnable()) {
1165 0 : return HCCL_SUCCESS;
1166 : }
1167 0 : CHK_PTR_NULL(groupname);
1168 0 : HCCL_INFO("[%s] START. thread [%llu], groupname[%s].", __func__, thread, groupname);
1169 :
1170 : DevType deviceType;
1171 0 : CHK_RET(hrtGetDeviceType(deviceType));
1172 0 : if (deviceType != DevType::DEV_TYPE_950 && deviceType != DevType::DEV_TYPE_960) {
1173 0 : return HCCL_SUCCESS;
1174 : }
1175 :
1176 0 : Thread *const threadPtr = reinterpret_cast<Thread*>(thread);
1177 0 : CHK_PRT_RET(threadPtr == nullptr, HCCL_ERROR("[%s] threadPtr is null", __func__), HCCL_E_PTR);
1178 0 : auto *const streamLitePtr = static_cast<Hccl::StreamLite *>(threadPtr->GetStreamLitePtr());
1179 0 : CHK_PRT_RET(streamLitePtr == nullptr, HCCL_ERROR("[%s] streamLitePtr is null", __func__), HCCL_E_PTR);
1180 : //FlagTaskInfo Report
1181 0 : Hccl::FlagTaskInfo flagTaskInfo;
1182 0 : flagTaskInfo.taskId = streamLitePtr->GetRtsq()->GetTaskId();
1183 0 : flagTaskInfo.type = Hccl::MainStreamTaskType::TAIL;
1184 :
1185 0 : Hccl::ProfilingHandlerLite::GetInstance().ReportMainStreamTask(flagTaskInfo);
1186 0 : return HCCL_SUCCESS;
1187 : }
|
