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 8 : inline HcclResult CheckDataTypeAndReduceOp(HcommDataType dataType, HcommReduceOp reduceOp)
383 : {
384 8 : 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 5 : 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 4 : 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 : HcclResult ret = HCCL_SUCCESS;
536 3 : if (threadPtr->IsDeviceA5()) {
537 3 : ret = CheckDataTypeAndReduceOp(dataType, reduceOp);
538 3 : CHK_PRT_RET(
539 : ret != HCCL_SUCCESS,
540 : HCCL_ERROR(
541 : "[%s] FAIL at CheckDataTypeAndReduceOp. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], "
542 : "count[%llu], dataType[%d], reduceOp[%d].",
543 : __func__, thread, channel, dst, src, count, dataType, reduceOp),
544 : ret);
545 : } else {
546 0 : CHK_PRT_RET(
547 : (IsSupportReduce(dataType, reduceOp) == false),
548 : HCCL_ERROR(
549 : "[%s] Not support reduce, "
550 : "dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d]",
551 : __func__, dst, src, count, dataType, reduceOp),
552 : HCCL_E_PARA);
553 : }
554 2 : uint64_t len = count * SIZE_TABLE[dataType];
555 :
556 2 : if (threadPtr->IsDeviceA5()) {
557 2 : auto* const transportLitePtr = reinterpret_cast<Hccl::BaseTransportLiteImpl*>(channel);
558 3 : CHK_PTR_NULL(transportLitePtr);
559 2 : auto* const streamLitePtr = static_cast<Hccl::StreamLite*>(threadPtr->GetStreamLitePtr());
560 2 : CHK_PTR_NULL(streamLitePtr);
561 :
562 2 : Hccl::RmaBufferLite locRmaBuf;
563 2 : ret = transportLitePtr->BuildLocRmaBufferLite(reinterpret_cast<uintptr_t>(src), len, locRmaBuf);
564 2 : CHK_PRT_RET(
565 : ret != HCCL_SUCCESS,
566 : HCCL_ERROR(
567 : "[%s] FAIL at BuildLocRmaBufferLite. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], "
568 : "count[%llu], dataType[%d], reduceOp[%d].",
569 : __func__, thread, channel, dst, src, count, dataType, reduceOp),
570 : ret);
571 1 : const Hccl::Buffer rmtBuf{reinterpret_cast<uintptr_t>(dst), len};
572 :
573 1 : Hccl::ReduceIn reduceIn{mapHcommDataTypeToA5.at(dataType), mapHcommReduceOpToA5.at(reduceOp)};
574 :
575 1 : EXCEPTION_CATCH(
576 : transportLitePtr->WriteReduce(locRmaBuf, rmtBuf, reduceIn, *streamLitePtr), ret = HCCL_E_INTERNAL);
577 1 : } else {
578 0 : HcclBuf locBuf{const_cast<void*>(src), len, nullptr};
579 0 : HcclBuf rmtBuf{dst, len, nullptr};
580 0 : HcclReduceInfo reduceInfo{static_cast<HcclDataType>(dataType), static_cast<HcclReduceOp>(reduceOp)};
581 :
582 0 : Stream* stream = GetStream(thread);
583 0 : CHK_PTR_NULL(stream);
584 :
585 0 : ret = HcclRemoteWriteReduce(stream, reinterpret_cast<void*>(channel), &rmtBuf, &locBuf, reduceInfo);
586 : }
587 1 : CHK_PRT_RET(
588 : ret != HCCL_SUCCESS,
589 : HCCL_ERROR(
590 : "[%s] FAIL. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], "
591 : "reduceOp[%d].",
592 : __func__, thread, channel, dst, src, count, dataType, reduceOp),
593 : ret);
594 1 : HCCL_INFO("[%s] SUCCESS.", __func__);
595 1 : return HCCL_SUCCESS;
596 : }
597 :
598 0 : HcclResult CommWriteReduceWithNotify(
599 : ThreadHandle thread, ChannelHandle channel, void* dst, const void* src, uint64_t count, HcommDataType dataType,
600 : HcommReduceOp reduceOp, uint32_t remoteNotifyIdx)
601 : {
602 0 : CHK_PTR_NULL(src);
603 0 : CHK_PTR_NULL(dst);
604 0 : CHK_RET(UnwrapChannelHandle(channel));
605 0 : AddThread(thread);
606 0 : CHK_PRT_RET(
607 : (IsSupportReduce(dataType, reduceOp) == false),
608 : HCCL_ERROR(
609 : "[%s] Not support reduce, "
610 : "dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d]",
611 : __func__, dst, src, count, dataType, reduceOp),
612 : HCCL_E_PARA);
613 0 : HcclBuf locBuf{const_cast<void*>(src), count * SIZE_TABLE[dataType], nullptr};
614 0 : HcclBuf rmtBuf{dst, count * SIZE_TABLE[dataType], nullptr};
615 0 : HcclReduceInfo reduceInfo{static_cast<HcclDataType>(dataType), static_cast<HcclReduceOp>(reduceOp)};
616 :
617 0 : Stream* stream = GetStream(thread);
618 0 : CHK_PTR_NULL(stream);
619 :
620 0 : return HcclRemoteWriteReduceWithNotify(
621 0 : stream, reinterpret_cast<void*>(channel), &rmtBuf, &locBuf, reduceInfo, remoteNotifyIdx);
622 : }
623 :
624 6 : int32_t HcommWriteWithNotifyOnThread(
625 : ThreadHandle thread, ChannelHandle channel, void* dst, const void* src, uint64_t len, uint32_t remoteNotifyIdx)
626 : {
627 6 : PLF_CONFIG_INFO(
628 : PLF_DATA_OP, "[%s] thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu], remoteNotifyIdx[%u].",
629 : __func__, thread, channel, dst, src, len, remoteNotifyIdx);
630 :
631 6 : CHK_PTR_NULL(dst);
632 5 : CHK_PTR_NULL(src);
633 4 : CHK_RET(UnwrapChannelHandle(channel));
634 3 : AddThread(thread);
635 :
636 3 : Thread* const threadPtr = reinterpret_cast<Thread*>(thread);
637 3 : CHK_PTR_NULL(threadPtr);
638 :
639 2 : HcclResult ret = HCCL_SUCCESS;
640 2 : if (threadPtr->IsDeviceA5()) {
641 2 : HCCL_DEBUG("[%s] Running on A5.", __func__);
642 2 : auto* const transportLitePtr = reinterpret_cast<Hccl::BaseTransportLiteImpl*>(channel);
643 3 : CHK_PTR_NULL(transportLitePtr);
644 2 : auto* const streamLitePtr = static_cast<Hccl::StreamLite*>(threadPtr->GetStreamLitePtr());
645 2 : CHK_PTR_NULL(streamLitePtr);
646 :
647 2 : Hccl::RmaBufferLite locRmaBuf;
648 2 : ret = transportLitePtr->BuildLocRmaBufferLite(reinterpret_cast<uintptr_t>(src), len, locRmaBuf);
649 2 : CHK_PRT_RET(
650 : ret != HCCL_SUCCESS,
651 : HCCL_ERROR(
652 : "[%s] FAIL at BuildLocRmaBufferLite. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], "
653 : "len[%llu], remoteNotifyIdx[%u].",
654 : __func__, thread, channel, dst, src, len, remoteNotifyIdx),
655 : ret);
656 1 : const Hccl::Buffer rmtBuf{reinterpret_cast<uintptr_t>(dst), len};
657 :
658 1 : Hccl::WithNotifyIn withNotify{Hccl::TransportNotifyType::NORMAL, remoteNotifyIdx};
659 :
660 1 : EXCEPTION_CATCH(
661 : transportLitePtr->WriteWithNotify(locRmaBuf, rmtBuf, withNotify, *streamLitePtr), ret = HCCL_E_INTERNAL);
662 1 : } else {
663 0 : HcclBuf locBuf{const_cast<void*>(src), len, nullptr};
664 0 : HcclBuf rmtBuf{dst, len, nullptr};
665 :
666 0 : Stream* stream = GetStream(thread);
667 0 : CHK_PTR_NULL(stream);
668 :
669 0 : ret = HcclRemoteWriteWithNotify(stream, reinterpret_cast<void*>(channel), &rmtBuf, &locBuf, remoteNotifyIdx);
670 : }
671 1 : CHK_PRT_RET(
672 : ret != HCCL_SUCCESS,
673 : HCCL_ERROR(
674 : "[%s] FAIL. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu], remoteNotifyIdx[%u].",
675 : __func__, thread, channel, dst, src, len, remoteNotifyIdx),
676 : ret);
677 1 : HCCL_INFO("[%s] SUCCESS.", __func__);
678 1 : return HCCL_SUCCESS;
679 : }
680 :
681 8 : int32_t HcommWriteReduceWithNotifyOnThread(
682 : ThreadHandle thread, ChannelHandle channel, void* dst, const void* src, uint64_t count, HcommDataType dataType,
683 : HcommReduceOp reduceOp, uint32_t remoteNotifyIdx)
684 : {
685 8 : CHK_PTR_NULL(dst);
686 7 : CHK_PTR_NULL(src);
687 6 : CHK_RET(UnwrapChannelHandle(channel));
688 :
689 5 : PLF_CONFIG_INFO(
690 : PLF_DATA_OP,
691 : "[%s] thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d], "
692 : "remoteNotifyIdx[%u].",
693 : __func__, thread, channel, dst, src, count, dataType, reduceOp, remoteNotifyIdx);
694 :
695 5 : AddThread(thread);
696 :
697 5 : Thread* const threadPtr = reinterpret_cast<Thread*>(thread);
698 5 : CHK_PTR_NULL(threadPtr);
699 :
700 4 : uint64_t len = count * SIZE_TABLE[dataType];
701 :
702 4 : HcclResult ret = HCCL_SUCCESS;
703 4 : if (threadPtr->IsDeviceA5()) {
704 4 : HCCL_DEBUG("[%s] Running on A5.", __func__);
705 4 : auto* const transportLitePtr = reinterpret_cast<Hccl::BaseTransportLiteImpl*>(channel);
706 7 : CHK_PTR_NULL(transportLitePtr);
707 4 : auto* const streamLitePtr = static_cast<Hccl::StreamLite*>(threadPtr->GetStreamLitePtr());
708 4 : CHK_PTR_NULL(streamLitePtr);
709 :
710 4 : Hccl::RmaBufferLite locRmaBuf;
711 4 : ret = transportLitePtr->BuildLocRmaBufferLite(reinterpret_cast<uintptr_t>(src), len, locRmaBuf);
712 4 : CHK_PRT_RET(
713 : ret != HCCL_SUCCESS,
714 : HCCL_ERROR(
715 : "[%s] FAIL at BuildLocRmaBufferLite. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], "
716 : "count[%llu], dataType[%d], reduceOp[%d], remoteNotifyIdx[%u].",
717 : __func__, thread, channel, dst, src, count, dataType, reduceOp, remoteNotifyIdx),
718 : ret);
719 3 : const Hccl::Buffer rmtBuf{reinterpret_cast<uintptr_t>(dst), len};
720 :
721 3 : ret = CheckDataTypeAndReduceOp(dataType, reduceOp);
722 3 : CHK_PRT_RET(
723 : ret != HCCL_SUCCESS,
724 : HCCL_ERROR(
725 : "[%s] FAIL at CheckDataTypeAndReduceOp. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], "
726 : "count[%llu], dataType[%d], reduceOp[%d], remoteNotifyIdx[%u].",
727 : __func__, thread, channel, dst, src, count, dataType, reduceOp, remoteNotifyIdx),
728 : ret);
729 1 : Hccl::ReduceIn reduceIn{mapHcommDataTypeToA5.at(dataType), mapHcommReduceOpToA5.at(reduceOp)};
730 :
731 1 : Hccl::WithNotifyIn withNotify{Hccl::TransportNotifyType::NORMAL, remoteNotifyIdx};
732 :
733 1 : EXCEPTION_CATCH(
734 : transportLitePtr->WriteReduceWithNotify(locRmaBuf, rmtBuf, reduceIn, withNotify, *streamLitePtr),
735 : ret = HCCL_E_INTERNAL);
736 3 : } else {
737 0 : ret = HCCL_E_NOT_SUPPORT;
738 : }
739 1 : CHK_PRT_RET(
740 : ret != HCCL_SUCCESS,
741 : HCCL_ERROR(
742 : "[%s] FAIL. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], "
743 : "reduceOp[%d], remoteNotifyIdx[%u].",
744 : __func__, thread, channel, dst, src, count, dataType, reduceOp, remoteNotifyIdx),
745 : ret);
746 1 : HCCL_INFO("[%s] SUCCESS.", __func__);
747 1 : return HCCL_SUCCESS;
748 : }
749 :
750 3 : int32_t HcommReadOnThread(ThreadHandle thread, ChannelHandle channel, void* dst, const void* src, uint64_t len)
751 : {
752 3 : CHK_PTR_NULL(dst);
753 3 : CHK_PTR_NULL(src);
754 3 : CHK_RET(UnwrapChannelHandle(channel));
755 :
756 3 : PLF_CONFIG_INFO(
757 : PLF_DATA_OP, "[%s] thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu].", __func__, thread,
758 : channel, dst, src, len);
759 :
760 3 : AddThread(thread);
761 :
762 3 : Thread* const threadPtr = reinterpret_cast<Thread*>(thread);
763 3 : CHK_PTR_NULL(threadPtr);
764 :
765 2 : HcclResult ret = HCCL_SUCCESS;
766 2 : if (threadPtr->IsDeviceA5()) {
767 2 : auto* const transportLitePtr = reinterpret_cast<Hccl::BaseTransportLiteImpl*>(channel);
768 3 : CHK_PTR_NULL(transportLitePtr);
769 2 : auto* const streamLitePtr = static_cast<Hccl::StreamLite*>(threadPtr->GetStreamLitePtr());
770 2 : CHK_PTR_NULL(streamLitePtr);
771 :
772 2 : Hccl::RmaBufferLite locRmaBuf;
773 2 : ret = transportLitePtr->BuildLocRmaBufferLite(reinterpret_cast<uintptr_t>(dst), len, locRmaBuf);
774 2 : CHK_PRT_RET(
775 : ret != HCCL_SUCCESS,
776 : HCCL_ERROR(
777 : "[%s] FAIL at BuildLocRmaBufferLite. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], "
778 : "len[%llu].",
779 : __func__, thread, channel, dst, src, len),
780 : ret);
781 1 : const Hccl::Buffer rmtBuf{reinterpret_cast<uintptr_t>(src), len};
782 :
783 1 : EXCEPTION_CATCH(transportLitePtr->Read(locRmaBuf, rmtBuf, *streamLitePtr), ret = HCCL_E_INTERNAL);
784 1 : } else {
785 0 : HcclBuf locBuf{dst, len, nullptr};
786 0 : HcclBuf rmtBuf{const_cast<void*>(src), len, nullptr};
787 :
788 0 : Stream* stream = GetStream(thread);
789 0 : CHK_PTR_NULL(stream);
790 :
791 0 : ret = HcclRemoteRead(stream, reinterpret_cast<void*>(channel), &locBuf, &rmtBuf);
792 : }
793 1 : CHK_PRT_RET(
794 : ret != HCCL_SUCCESS,
795 : HCCL_ERROR(
796 : "[%s] FAIL. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu].", __func__, thread,
797 : channel, dst, src, len),
798 : ret);
799 1 : HCCL_INFO("[%s] SUCCESS.", __func__);
800 1 : return HCCL_SUCCESS;
801 : }
802 :
803 4 : int32_t HcommReadReduceOnThread(
804 : ThreadHandle thread, ChannelHandle channel, void* dst, const void* src, uint64_t count, HcommDataType dataType,
805 : HcommReduceOp reduceOp)
806 : {
807 4 : CHK_PTR_NULL(dst);
808 4 : CHK_PTR_NULL(src);
809 4 : CHK_RET(UnwrapChannelHandle(channel));
810 :
811 4 : PLF_CONFIG_INFO(
812 : PLF_DATA_OP,
813 : "[%s] thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d].",
814 : __func__, thread, channel, dst, src, count, dataType, reduceOp);
815 :
816 4 : AddThread(thread);
817 :
818 4 : Thread* const threadPtr = reinterpret_cast<Thread*>(thread);
819 4 : CHK_PTR_NULL(threadPtr);
820 :
821 3 : uint64_t len = count * SIZE_TABLE[dataType];
822 :
823 3 : HcclResult ret = HCCL_SUCCESS;
824 3 : if (threadPtr->IsDeviceA5()) {
825 3 : auto* const transportLitePtr = reinterpret_cast<Hccl::BaseTransportLiteImpl*>(channel);
826 5 : CHK_PTR_NULL(transportLitePtr);
827 3 : auto* const streamLitePtr = static_cast<Hccl::StreamLite*>(threadPtr->GetStreamLitePtr());
828 3 : CHK_PTR_NULL(streamLitePtr);
829 :
830 3 : Hccl::RmaBufferLite locRmaBuf;
831 3 : ret = transportLitePtr->BuildLocRmaBufferLite(reinterpret_cast<uintptr_t>(dst), len, locRmaBuf);
832 3 : CHK_PRT_RET(
833 : ret != HCCL_SUCCESS,
834 : HCCL_ERROR(
835 : "[%s] FAIL at BuildLocRmaBufferLite. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], "
836 : "count[%llu], dataType[%d], reduceOp[%d].",
837 : __func__, thread, channel, dst, src, count, dataType, reduceOp),
838 : ret);
839 2 : const Hccl::Buffer rmtBuf{reinterpret_cast<uintptr_t>(src), len};
840 :
841 2 : ret = CheckDataTypeAndReduceOp(dataType, reduceOp);
842 2 : CHK_PRT_RET(
843 : ret != HCCL_SUCCESS,
844 : HCCL_ERROR(
845 : "[%s] FAIL at CheckDataTypeAndReduceOp. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], "
846 : "count[%llu], dataType[%d], reduceOp[%d].",
847 : __func__, thread, channel, dst, src, count, dataType, reduceOp),
848 : ret);
849 1 : Hccl::ReduceIn reduceIn{mapHcommDataTypeToA5.at(dataType), mapHcommReduceOpToA5.at(reduceOp)};
850 :
851 1 : EXCEPTION_CATCH(
852 : transportLitePtr->ReadReduce(locRmaBuf, rmtBuf, reduceIn, *streamLitePtr), ret = HCCL_E_INTERNAL);
853 2 : } else {
854 0 : CHK_PRT_RET(
855 : (IsSupportReduce(dataType, reduceOp) == false),
856 : HCCL_ERROR(
857 : "[%s] Not support reduce, "
858 : "dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d]",
859 : __func__, dst, src, count, dataType, reduceOp),
860 : HCCL_E_PARA);
861 0 : HcclBuf locBuf{dst, len, nullptr};
862 0 : HcclBuf rmtBuf{const_cast<void*>(src), len, nullptr};
863 0 : HcclReduceInfo reduceInfo{static_cast<HcclDataType>(dataType), static_cast<HcclReduceOp>(reduceOp)};
864 :
865 0 : Stream* stream = GetStream(thread);
866 0 : CHK_PTR_NULL(stream);
867 :
868 0 : ret = HcclRemoteReadReduce(stream, reinterpret_cast<void*>(channel), &locBuf, &rmtBuf, reduceInfo);
869 : }
870 1 : CHK_PRT_RET(
871 : ret != HCCL_SUCCESS,
872 : HCCL_ERROR(
873 : "[%s] FAIL. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], "
874 : "reduceOp[%d].",
875 : __func__, thread, channel, dst, src, count, dataType, reduceOp),
876 : ret);
877 1 : HCCL_INFO("[%s] SUCCESS.", __func__);
878 1 : return HCCL_SUCCESS;
879 : }
880 :
881 16 : int32_t HcommBatchTransferOnThread(
882 : ThreadHandle thread, ChannelHandle channel, const HcommBatchTransferDesc* transferDescs, uint32_t transferDescNum)
883 : {
884 16 : CHK_PTR_NULL(transferDescs);
885 15 : CHK_RET(UnwrapChannelHandle(channel));
886 :
887 14 : PLF_CONFIG_INFO(
888 : PLF_DATA_OP, "[%s] thread[0x%llx], channel[0x%llx], transferDescNum[%u].", __func__, thread, channel,
889 : transferDescNum);
890 :
891 14 : CHK_PRT_RET(transferDescNum == 0, HCCL_ERROR("[%s] transferDescNum is 0.", __func__), HCCL_E_PARA);
892 :
893 13 : Thread* const threadPtr = reinterpret_cast<Thread*>(thread);
894 13 : CHK_PTR_NULL(threadPtr);
895 12 : AddThread(thread);
896 12 : HcclResult ret = HCCL_SUCCESS;
897 12 : if (threadPtr->IsDeviceA5()) {
898 12 : auto* const ubTransportLitePtr = reinterpret_cast<Hccl::UbTransportLiteImpl*>(channel);
899 12 : CHK_PTR_NULL(ubTransportLitePtr);
900 12 : auto* const streamLitePtr = static_cast<Hccl::StreamLite*>(threadPtr->GetStreamLitePtr());
901 12 : CHK_PTR_NULL(streamLitePtr);
902 12 : ret = ubTransportLitePtr->ExecuteBatchTransfer(streamLitePtr, transferDescs, transferDescNum);
903 : } else {
904 0 : Stream* stream = GetStream(thread);
905 0 : CHK_PTR_NULL(stream);
906 0 : hccl::Transport* transport = reinterpret_cast<hccl::Transport*>(channel);
907 0 : CHK_PTR_NULL(transport);
908 0 : ret = transport->BatchTransferAsync(transferDescs, transferDescNum, *stream);
909 0 : if (ret == HCCL_E_NOT_SUPPORT) {
910 0 : return HCCL_E_NOT_SUPPORT;
911 : }
912 : }
913 12 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[%s] BatchTransferAsync failed.", __func__), ret);
914 :
915 8 : HCCL_INFO("[%s] SUCCESS. transferDescNum[%u].", __func__, transferDescNum);
916 8 : return ret;
917 : }
918 :
919 0 : int32_t HcommWriteNbiOnThread(ThreadHandle thread, ChannelHandle channel, void* dst, const void* src, uint64_t len)
920 : {
921 0 : HCCL_DEBUG(
922 : "[%s] thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu].", __func__, thread, channel, dst,
923 : src, len);
924 0 : CHK_PTR_NULL(src);
925 0 : CHK_PTR_NULL(dst);
926 0 : return HCCL_E_NOT_SUPPORT;
927 : }
928 :
929 0 : int32_t HcommWriteNbi(ChannelHandle channel, void* dst, const void* src, uint64_t len)
930 : {
931 0 : HCCL_DEBUG("[%s] channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu].", __func__, channel, dst, src, len);
932 0 : CHK_PTR_NULL(src);
933 0 : CHK_PTR_NULL(dst);
934 0 : return HCCL_E_NOT_SUPPORT;
935 : }
936 :
937 0 : int32_t HcommWriteWithNotifyNbiOnThread(
938 : ThreadHandle thread, ChannelHandle channel, void* dst, const void* src, uint64_t len, uint32_t remoteNotifyIdx)
939 : {
940 0 : HCCL_DEBUG(
941 : "[%s] thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu], remoteNotifyIdx[%u].", __func__,
942 : thread, channel, dst, src, len, remoteNotifyIdx);
943 0 : CHK_PTR_NULL(src);
944 0 : CHK_PTR_NULL(dst);
945 0 : return HCCL_E_NOT_SUPPORT;
946 : }
947 :
948 : int32_t
949 0 : HcommWriteWithNotifyNbi(ChannelHandle channel, void* dst, const void* src, uint64_t len, uint32_t remoteNotifyIdx)
950 : {
951 0 : HCCL_DEBUG(
952 : "[%s] channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu], remoteNotifyIdx[%u].", __func__, channel, dst, src,
953 : len, remoteNotifyIdx);
954 0 : CHK_PTR_NULL(src);
955 0 : CHK_PTR_NULL(dst);
956 0 : return HCCL_E_NOT_SUPPORT;
957 : }
958 :
959 0 : int32_t HcommReadNbiOnThread(ThreadHandle thread, ChannelHandle channel, void* dst, const void* src, uint64_t len)
960 : {
961 0 : HCCL_DEBUG(
962 : "[%s] thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu].", __func__, thread, channel, dst,
963 : src, len);
964 0 : CHK_PTR_NULL(src);
965 0 : CHK_PTR_NULL(dst);
966 0 : return HCCL_E_NOT_SUPPORT;
967 : }
968 :
969 0 : int32_t HcommReadNbi(ChannelHandle channel, void* dst, const void* src, uint64_t len)
970 : {
971 0 : HCCL_DEBUG("[%s] channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu].", __func__, channel, dst, src, len);
972 0 : CHK_PTR_NULL(src);
973 0 : CHK_PTR_NULL(dst);
974 0 : return HCCL_E_NOT_SUPPORT;
975 : }
976 :
977 3 : int32_t HcommChannelNotifyRecordOnThread(ThreadHandle thread, ChannelHandle channel, uint32_t remoteNotifyIdx)
978 : {
979 3 : CHK_RET(UnwrapChannelHandle(channel));
980 :
981 2 : PLF_CONFIG_INFO(
982 : PLF_DATA_OP, "[%s] thread[0x%llx], channel[0x%llx], remoteNotifyIdx[%u].", __func__, thread, channel,
983 : remoteNotifyIdx);
984 :
985 2 : AddThread(thread);
986 :
987 2 : Thread* const threadPtr = reinterpret_cast<Thread*>(thread);
988 2 : CHK_PTR_NULL(threadPtr);
989 :
990 1 : HcclResult ret = HCCL_SUCCESS;
991 1 : if (threadPtr->IsDeviceA5()) {
992 1 : HCCL_DEBUG("[%s] Running on A5.", __func__);
993 1 : auto* const transportLitePtr = reinterpret_cast<Hccl::BaseTransportLiteImpl*>(channel);
994 1 : CHK_PTR_NULL(transportLitePtr);
995 1 : auto* const streamLitePtr = static_cast<Hccl::StreamLite*>(threadPtr->GetStreamLitePtr());
996 1 : CHK_PTR_NULL(streamLitePtr);
997 1 : HCCL_INFO("channel streamlite ptr %p.", streamLitePtr);
998 :
999 1 : EXCEPTION_CATCH(transportLitePtr->Post(remoteNotifyIdx, *streamLitePtr), ret = HCCL_E_INTERNAL);
1000 : } else {
1001 0 : Stream* stream = GetStream(thread);
1002 0 : CHK_PTR_NULL(stream);
1003 :
1004 0 : ret = HcclRemoteNotifyRecord(stream, reinterpret_cast<void*>(channel), remoteNotifyIdx);
1005 : }
1006 1 : CHK_PRT_RET(
1007 : ret != HCCL_SUCCESS,
1008 : HCCL_ERROR(
1009 : "[%s] FAIL. thread[0x%llx], channel[0x%llx], remoteNotifyIdx[%u].", __func__, thread, channel,
1010 : remoteNotifyIdx),
1011 : ret);
1012 1 : HCCL_INFO("[%s] SUCCESS.", __func__);
1013 1 : return HCCL_SUCCESS;
1014 : }
1015 :
1016 0 : int32_t HcommChannelNotifyRecord(ChannelHandle channel, uint32_t remoteNotifyIdx)
1017 : {
1018 0 : HCCL_DEBUG("[%s] channel[0x%llx], remoteNotifyIdx[%u].", __func__, channel, remoteNotifyIdx);
1019 0 : return HCCL_E_NOT_SUPPORT;
1020 : }
1021 :
1022 : int32_t
1023 10 : HcommChannelNotifyWaitOnThread(ThreadHandle thread, ChannelHandle channel, uint32_t localNotifyIdx, uint32_t timeOut)
1024 : {
1025 10 : CHK_RET(UnwrapChannelHandle(channel));
1026 :
1027 8 : PLF_CONFIG_INFO(
1028 : PLF_DATA_OP, "[%s] thread[0x%llx], channel[0x%llx], localNotifyIdx[%u], timeOut[%u s].", __func__, thread,
1029 : channel, localNotifyIdx, timeOut);
1030 :
1031 8 : AddThread(thread);
1032 :
1033 8 : Thread* const threadPtr = reinterpret_cast<Thread*>(thread);
1034 8 : CHK_PTR_NULL(threadPtr);
1035 :
1036 6 : HcclResult ret = HCCL_SUCCESS;
1037 6 : if (threadPtr->IsDeviceA5()) {
1038 6 : HCCL_DEBUG("[%s] Running on A5.", __func__);
1039 6 : auto* const transportLitePtr = reinterpret_cast<Hccl::BaseTransportLiteImpl*>(channel);
1040 6 : CHK_PTR_NULL(transportLitePtr);
1041 6 : auto* const streamLitePtr = static_cast<Hccl::StreamLite*>(threadPtr->GetStreamLitePtr());
1042 6 : CHK_PTR_NULL(streamLitePtr);
1043 :
1044 6 : EXCEPTION_CATCH(
1045 : transportLitePtr->WaitWithTimeout(localNotifyIdx, *streamLitePtr, timeOut), ret = HCCL_E_INTERNAL);
1046 : } else {
1047 0 : Stream* stream = GetStream(thread);
1048 0 : CHK_PTR_NULL(stream);
1049 :
1050 0 : ret = HcclRemoteNotifyWait(stream, reinterpret_cast<void*>(channel), localNotifyIdx, timeOut);
1051 : }
1052 6 : CHK_PRT_RET(
1053 : ret != HCCL_SUCCESS,
1054 : HCCL_ERROR(
1055 : "[%s] FAIL. thread[0x%llx], channel[0x%llx], localNotifyIdx[%u], timeOut[%u s].", __func__, thread, channel,
1056 : localNotifyIdx, timeOut),
1057 : ret);
1058 6 : HCCL_INFO("[%s] SUCCESS.", __func__);
1059 6 : return HCCL_SUCCESS;
1060 : }
1061 :
1062 0 : int32_t HcommChannelNotifyWait(ChannelHandle channel, uint32_t localNotifyIdx, uint32_t timeOut)
1063 : {
1064 0 : HCCL_DEBUG("[%s] channel[0x%llx], localNotifyIdx[%u], timeOut[%u s].", __func__, channel, localNotifyIdx, timeOut);
1065 0 : return HCCL_E_NOT_SUPPORT;
1066 : }
1067 :
1068 0 : HcclResult CommFence(ThreadHandle thread, ChannelHandle channel) // 控制前后的任务保序
1069 : {
1070 0 : CHK_RET(UnwrapChannelHandle(channel));
1071 :
1072 0 : HCCL_DEBUG("[CommFence] thread[0x%llx], channel[0x%llx].", thread, channel);
1073 0 : Stream* stream = GetStream(thread);
1074 0 : CHK_PTR_NULL(stream);
1075 :
1076 0 : return HcclRemoteFence(stream, reinterpret_cast<void*>(channel), false);
1077 : }
1078 :
1079 0 : int32_t HcommSetLaunchMode(const char* launchTag, HcommLaunchMode mode)
1080 : {
1081 0 : HCCL_DEBUG("HcommSetLaunchMode launchTag[%s]", launchTag);
1082 0 : return g_threadLaunchCtx.SetLaunchMode(launchTag, mode);
1083 : }
1084 :
1085 0 : int32_t HcommBatchModeStart(const char* batchTag) { return HcommSetLaunchMode(batchTag, HCOMM_LAUNCH_MODE_BATCH); }
1086 :
1087 0 : int32_t HcommBatchModeEnd(const char* batchTag) { return HcommSetLaunchMode(batchTag, HCOMM_LAUNCH_MODE_EAGER); }
1088 :
1089 0 : int32_t HcommAcquireComm(const char* commId)
1090 : {
1091 0 : CHK_PTR_NULL(commId);
1092 : DevType deviceType;
1093 0 : CHK_RET(hrtGetDeviceType(deviceType));
1094 0 : HCCL_INFO("[%s]comId[%s], devType[%d]", __func__, commId, deviceType);
1095 0 : if (deviceType != DevType::DEV_TYPE_950 && deviceType != DevType::DEV_TYPE_960) {
1096 0 : HcclCommAicpu* hcclComm = AicpuHcclProcess::AicpuGetCommbyGroup(commId);
1097 0 : CHK_PRT_RET(!hcclComm, HCCL_ERROR("%s AicpuGetCommbyGroup is null, commId[%s]", __func__, commId), HCCL_E_PTR);
1098 0 : CHK_RET(hcclComm->SetDispatcherCtxOnThread());
1099 0 : } else {
1100 0 : CollCommAicpu* hcclComm = CollCommAicpuMgr::GetInstance().AcquireCommForUse(commId);
1101 0 : CHK_PRT_RET(!hcclComm, HCCL_ERROR("%s AcquireCommForUse is null, commId[%s]", __func__, commId), HCCL_E_PTR);
1102 : }
1103 0 : return HCCL_SUCCESS;
1104 : }
1105 :
1106 0 : int32_t HcommChannelRegisterDfx(
1107 : ChannelHandle channel, [[maybe_unused]] std::function<HcclResult(u32, u32, const Hccl::TaskParam&, u64)> callback)
1108 : {
1109 0 : CHK_RET(UnwrapChannelHandle(channel));
1110 0 : HCCL_INFO("[HcommChannelRegisterDfx] ChannelHandle[0x%llx] Init success", channel);
1111 0 : return HCCL_SUCCESS;
1112 : }
1113 :
1114 : int32_t
1115 0 : HcommThreadRegisterDfx(ThreadHandle thread, std::function<HcclResult(u32, u32, const Hccl::TaskParam&, u64)> callback)
1116 : {
1117 0 : Thread* threadPtr = reinterpret_cast<Thread*>(thread);
1118 0 : CHK_PTR_NULL(threadPtr);
1119 0 : CHK_RET(threadPtr->SetAddTaskInfoCallback(callback));
1120 0 : HCCL_INFO("[HcommThreadRegisterDfx] ThreadHandle[0x%llx] Init success", thread);
1121 0 : return HCCL_SUCCESS;
1122 : }
1123 :
1124 0 : int32_t HcommThreadRegisterCheckExecStatus(ThreadHandle thread, std::function<HcclResult(bool)> callback)
1125 : {
1126 0 : Thread* threadPtr = reinterpret_cast<Thread*>(thread);
1127 0 : CHK_PTR_NULL(threadPtr);
1128 0 : CHK_RET(threadPtr->SetCheckExecStatusCallback(callback));
1129 0 : return HCCL_SUCCESS;
1130 : }
1131 :
1132 0 : int32_t HcommNewThreadRegisterDfx(ThreadHandle thread, std::function<void(Hccl::TaskInfoCircularQueue*)> callback)
1133 : {
1134 0 : hccl::AicpuTsThread* tsThread = reinterpret_cast<hccl::AicpuTsThread*>(thread);
1135 0 : CHK_PTR_NULL(tsThread);
1136 0 : tsThread->SetReportStreamTaskCallback(std::move(callback));
1137 0 : HCCL_INFO("[HcommNewThreadRegisterDfx] ThreadHandle[0x%llx] Init success", thread);
1138 0 : return HCCL_SUCCESS;
1139 : }
1140 :
1141 0 : int32_t HcommNewThreadRegisterGetLatestDfxOpInfo(ThreadHandle thread, std::function<const void*()> callback)
1142 : {
1143 0 : hccl::AicpuTsThread* tsThread = reinterpret_cast<hccl::AicpuTsThread*>(thread);
1144 0 : CHK_PTR_NULL(tsThread);
1145 0 : tsThread->SetGetLatestDfxOpInfoCallback(std::move(callback));
1146 0 : HCCL_INFO("[HcommNewThreadRegisterGetLatestDfxOpInfo] ThreadHandle[0x%llx] Init success", thread);
1147 0 : return HCCL_SUCCESS;
1148 : }
1149 :
1150 0 : int32_t HcommReleaseComm(const char* commId)
1151 : {
1152 0 : CHK_PTR_NULL(commId);
1153 : DevType deviceType;
1154 0 : CHK_RET(hrtGetDeviceType(deviceType));
1155 0 : HCCL_INFO("[%s]comId[%s], devType[%d]", __func__, commId, deviceType);
1156 0 : if (deviceType != DevType::DEV_TYPE_950 && deviceType != DevType::DEV_TYPE_960) {
1157 0 : AicpuHcclProcess::AicpuReleaseCommbyGroup(commId);
1158 : } else {
1159 0 : CollCommAicpuMgr::GetInstance().ReleaseComm(commId);
1160 : }
1161 0 : return HCCL_SUCCESS;
1162 : }
1163 :
1164 0 : int32_t HcommFenceOnThread(ThreadHandle thread)
1165 : {
1166 0 : HCCL_DEBUG("[%s] thread[0x%llx].", __func__, thread);
1167 0 : return HCCL_E_NOT_SUPPORT;
1168 : }
1169 :
1170 : #ifdef __cplusplus
1171 : extern "C" {
1172 : #endif // __cplusplus
1173 0 : int32_t HcommFlush() { return HCCL_E_NOT_SUPPORT; }
1174 :
1175 3 : int32_t HcommChannelFenceOnThread(ThreadHandle thread, ChannelHandle channel)
1176 : {
1177 3 : CHK_RET(UnwrapChannelHandle(channel));
1178 :
1179 2 : HCCL_DEBUG("[%s] thread[0x%llx], channel[0x%llx].", __func__, thread, channel);
1180 2 : Thread* const threadPtr = reinterpret_cast<Thread*>(thread);
1181 2 : CHK_PTR_NULL(threadPtr);
1182 1 : if (threadPtr->IsDeviceA5()) {
1183 1 : auto* const transportLitePtr = reinterpret_cast<Hccl::BaseTransportLiteImpl*>(channel);
1184 1 : CHK_PTR_NULL(transportLitePtr);
1185 1 : CHK_RET(transportLitePtr->Fence());
1186 : }
1187 :
1188 1 : return HCCL_SUCCESS;
1189 : }
1190 :
1191 0 : int32_t HcommChannelFence(ChannelHandle channel)
1192 : {
1193 0 : HCCL_DEBUG("[%s] channel[0x%llx].", __func__, channel);
1194 0 : return HCCL_E_NOT_SUPPORT;
1195 : }
1196 :
1197 0 : int32_t HcommThreadJoin(ThreadHandle thread, uint32_t timeout)
1198 : {
1199 0 : hccl::Thread* threadPtr = reinterpret_cast<hccl::Thread*>(thread);
1200 0 : CHK_PTR_NULL(threadPtr);
1201 :
1202 0 : HCCL_INFO("[%s] START. thread[0x%llx].", __func__, thread);
1203 :
1204 0 : if (threadPtr->IsDeviceA5()) {
1205 0 : HCCL_INFO("[%s] Running on A5.", __func__);
1206 0 : auto* const streamLitePtr = static_cast<Hccl::StreamLite*>(threadPtr->GetStreamLitePtr());
1207 0 : CHK_PTR_NULL(streamLitePtr);
1208 0 : auto* const rtsqPtr = streamLitePtr->GetRtsq();
1209 0 : CHK_PTR_NULL(rtsqPtr);
1210 :
1211 0 : uint32_t head = 0;
1212 0 : uint32_t tail = 0;
1213 0 : uint32_t sqId = streamLitePtr->GetSqId();
1214 0 : EXCEPTION_CATCH(tail = rtsqPtr->QuerySqTail(), return HCCL_E_INTERNAL);
1215 0 : HCCL_INFO("[%s] aicpu stream sqid[%u] tail[%u]", __func__, sqId, tail);
1216 :
1217 0 : u64 startUsec = GetCurAicpuTimestamp();
1218 0 : u64 lastUsec = startUsec;
1219 0 : constexpr uint64_t NANOSECOND_TO_SECOND = 1000000000U;
1220 0 : const uint64_t kPrintSqInterval = 30U;
1221 : do {
1222 0 : EXCEPTION_CATCH(head = rtsqPtr->QuerySqHead(), return HCCL_E_INTERNAL);
1223 0 : u64 curUsec = GetCurAicpuTimestamp();
1224 0 : if (curUsec - startUsec > NANOSECOND_TO_SECOND * timeout) {
1225 0 : HCCL_ERROR("[%s] timeout %us. curhead:%u, curtail:%u, sqId:%u", __func__, timeout, head, tail, sqId);
1226 0 : return HCCL_E_TIMEOUT;
1227 : }
1228 :
1229 : // 等待下发阶段，每隔30s打印一次状态
1230 0 : if (curUsec - lastUsec > NANOSECOND_TO_SECOND * kPrintSqInterval) {
1231 0 : lastUsec = curUsec;
1232 0 : HCCL_RUN_INFO("[%s]Current state. sqid:%d, head:%u, tail:%u", __func__, sqId, head, tail);
1233 : }
1234 0 : } while (head != tail);
1235 0 : HCCL_INFO("[%s] SUCCESS. RTSQ's head[%u] == tail[%u].", __func__, head, tail);
1236 0 : return HCCL_SUCCESS;
1237 : }
1238 :
1239 0 : HCCL_ERROR("[%s]Does not support this interface.", __func__);
1240 0 : return HCCL_E_NOT_SUPPORT;
1241 : }
1242 :
1243 0 : int32_t HcommChannelDrainOnThread(ThreadHandle thread, ChannelHandle channel)
1244 : {
1245 0 : CHK_RET(UnwrapChannelHandle(channel));
1246 :
1247 0 : AddThread(thread);
1248 0 : Thread* const threadPtr = reinterpret_cast<Thread*>(thread);
1249 0 : CHK_PTR_NULL(threadPtr);
1250 0 : Stream* stream = GetStream(thread);
1251 0 : CHK_PTR_NULL(stream);
1252 :
1253 0 : HcclResult ret = HCCL_SUCCESS;
1254 0 : if (threadPtr->IsDeviceA5()) {
1255 0 : HCCL_DEBUG("[%s] Running on A5.", __func__);
1256 0 : auto* const transportLitePtr = reinterpret_cast<Hccl::BaseTransportLiteImpl*>(channel);
1257 0 : CHK_PTR_NULL(transportLitePtr);
1258 0 : auto* const streamLitePtr = static_cast<Hccl::StreamLite*>(threadPtr->GetStreamLitePtr());
1259 0 : CHK_PTR_NULL(streamLitePtr);
1260 :
1261 0 : EXCEPTION_CATCH(transportLitePtr->Drain(*streamLitePtr), ret = HCCL_E_INTERNAL);
1262 0 : return ret;
1263 : }
1264 :
1265 0 : ret = HcclRemoteDrain(stream, reinterpret_cast<void*>(channel));
1266 0 : CHK_PRT_RET(
1267 : ret != HCCL_SUCCESS, HCCL_ERROR("[%s] Run FAIL. thread[0x%llx], channel[0x%llx].", __func__, thread, channel),
1268 : ret);
1269 :
1270 0 : return HCCL_SUCCESS;
1271 : }
1272 : #ifdef __cplusplus
1273 : }
1274 : #endif // __cplusplus
1275 :
1276 0 : HcclResult HcommProfilingReportDeviceOp(const char* groupname)
1277 : {
1278 0 : if (!GetProfilingEnable()) {
1279 0 : return HCCL_SUCCESS;
1280 : }
1281 0 : CHK_PTR_NULL(groupname);
1282 :
1283 : DevType deviceType;
1284 0 : CHK_RET(hrtGetDeviceType(deviceType));
1285 0 : if (deviceType != DevType::DEV_TYPE_950 && deviceType != DevType::DEV_TYPE_960) {
1286 0 : return HCCL_SUCCESS;
1287 : }
1288 :
1289 0 : CollCommAicpu* currentComm = CollCommAicpuMgr::GetInstance().GetCurrentComm();
1290 0 : CHK_PTR_NULL(currentComm);
1291 0 : CHK_RET(currentComm->ProfilingReportDeviceOp());
1292 0 : return HCCL_SUCCESS;
1293 : }
1294 :
1295 0 : HcclResult HcommProfilingReportKernelStartTask(uint64_t thread, const char* groupname)
1296 : {
1297 0 : if (!GetProfilingEnable()) {
1298 0 : return HCCL_SUCCESS;
1299 : }
1300 :
1301 : DevType deviceType;
1302 0 : CHK_RET(hrtGetDeviceType(deviceType));
1303 0 : if (deviceType != DevType::DEV_TYPE_950 && deviceType != DevType::DEV_TYPE_960) {
1304 0 : return HCCL_SUCCESS;
1305 : }
1306 0 : CHK_PTR_NULL(groupname);
1307 0 : CollCommAicpu* currentComm = CollCommAicpuMgr::GetInstance().GetCurrentComm();
1308 0 : CHK_PTR_NULL(currentComm);
1309 0 : CHK_RET(currentComm->UpdateTask());
1310 0 : Thread* const threadPtr = reinterpret_cast<Thread*>(thread);
1311 0 : CHK_PTR_NULL(threadPtr);
1312 0 : auto* const streamLitePtr = static_cast<Hccl::StreamLite*>(threadPtr->GetStreamLitePtr());
1313 0 : CHK_PTR_NULL(streamLitePtr);
1314 : Hccl::DfxFlagTaskInfo flagTaskInfo;
1315 0 : flagTaskInfo.taskId = streamLitePtr->GetRtsq()->GetTaskId();
1316 0 : flagTaskInfo.type = Hccl::DfxMainStreamTaskType::HEAD;
1317 0 : Hccl::DfxProfilingHandlerLite::GetInstance().ReportMainStreamTask(flagTaskInfo);
1318 0 : HCCL_INFO("[%s] END, thread [%llu], groupname[%s], taskId[%u].", __func__, thread, groupname, flagTaskInfo.taskId);
1319 0 : return HCCL_SUCCESS;
1320 : }
1321 :
1322 0 : HcclResult HcommProfilingReportKernelEndTask(uint64_t thread, const char* groupname)
1323 : {
1324 0 : if (!GetProfilingEnable()) {
1325 0 : return HCCL_SUCCESS;
1326 : }
1327 0 : CHK_PTR_NULL(groupname);
1328 0 : HCCL_INFO("[%s] START. thread [%llu], groupname[%s].", __func__, thread, groupname);
1329 :
1330 : DevType deviceType;
1331 0 : CHK_RET(hrtGetDeviceType(deviceType));
1332 0 : if (deviceType != DevType::DEV_TYPE_950 && deviceType != DevType::DEV_TYPE_960) {
1333 0 : return HCCL_SUCCESS;
1334 : }
1335 :
1336 0 : Thread* const threadPtr = reinterpret_cast<Thread*>(thread);
1337 0 : CHK_PRT_RET(threadPtr == nullptr, HCCL_ERROR("[%s] threadPtr is null", __func__), HCCL_E_PTR);
1338 0 : auto* const streamLitePtr = static_cast<Hccl::StreamLite*>(threadPtr->GetStreamLitePtr());
1339 0 : CHK_PRT_RET(streamLitePtr == nullptr, HCCL_ERROR("[%s] streamLitePtr is null", __func__), HCCL_E_PTR);
1340 : // FlagTaskInfo Report
1341 : Hccl::DfxFlagTaskInfo flagTaskInfo;
1342 0 : flagTaskInfo.type = Hccl::DfxMainStreamTaskType::TAIL;
1343 0 : auto* rtsq = streamLitePtr->GetRtsq();
1344 0 : CHK_PRT_RET(rtsq == nullptr, HCCL_ERROR("[%s] rtsq is null", __func__), HCCL_E_PTR);
1345 0 : uint16_t streamId = 0;
1346 0 : uint16_t taskId = 0;
1347 0 : HcclResult ret = rtsq->GetLastStreamIdAndTaskId(streamId, taskId);
1348 0 : CHK_PRT_RET(
1349 : ret != HCCL_SUCCESS,
1350 : HCCL_ERROR("[%s] GetLastStreamIdAndTaskId fail, ret[%d], sqId[%u].", __func__, ret, streamLitePtr->GetSqId()),
1351 : ret);
1352 0 : constexpr uint32_t UINT16_BIT_WIDTH = 16U;
1353 0 : flagTaskInfo.taskId = (static_cast<uint32_t>(taskId) << UINT16_BIT_WIDTH) | static_cast<uint32_t>(streamId);
1354 :
1355 0 : Hccl::DfxProfilingHandlerLite::GetInstance().ReportMainStreamTask(flagTaskInfo);
1356 0 : return HCCL_SUCCESS;
1357 : }
|
