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