Line data Source code
1 : /**
2 : * Copyright (c) 2025 Huawei Technologies Co., Ltd.
3 : * This program is free software, you can redistribute it and/or modify it under the terms and conditions of
4 : * CANN Open Software License Agreement Version 2.0 (the "License").
5 : * Please refer to the License for details. You may not use this file except in compliance with the License.
6 : * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,
7 : * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
8 : * See LICENSE in the root of the software repository for the full text of the License.
9 : */
10 :
11 : #include "hccl_api_data.h"
12 : #include "new/hccl_primitive_local.h"
13 : #include "new/hccl_primitive_remote.h"
14 : #include "thread.h"
15 : #include "launch_context.h"
16 : #include "op_base_v2.h"
17 : #include "host/host_cpu_roce_channel.h"
18 : #include "hccl_comm_pub.h"
19 : #include "op_base.h"
20 : #include "dlprof_function.h"
21 : #include "hcclCommOp.h"
22 : #include "adapter_prof.h"
23 : #include "hccl_diag.h"
24 : #include "exception_handler.h"
25 : #include "task_info.h"
26 : #include "task_param.h"
27 : #ifdef ENABLE_EXPERIMENTAL
28 : #include "nic_plugin_dispatcher.h"
29 : #endif
30 :
31 : using namespace hccl;
32 : thread_local LaunchContext g_threadLaunchCtx;
33 :
34 9 : void AddThreadWithTag(ThreadHandle thread) {
35 9 : g_threadLaunchCtx.AddThreadWithTag(thread);
36 9 : }
37 :
38 0 : bool IsSupportReduce(HcommDataType dataType, HcommReduceOp op)
39 : {
40 0 : bool checkDataType =
41 0 : (dataType == HCOMM_DATA_TYPE_FP32 || dataType == HCOMM_DATA_TYPE_FP16 || dataType == HCOMM_DATA_TYPE_INT8 ||
42 0 : dataType == HCOMM_DATA_TYPE_INT16 || dataType == HCOMM_DATA_TYPE_INT32 || dataType == HCOMM_DATA_TYPE_BFP16);
43 0 : bool checkReduceType = (op == HCOMM_REDUCE_SUM || op == HCOMM_REDUCE_MAX || op == HCOMM_REDUCE_MIN);
44 0 : return checkDataType && checkReduceType;
45 : }
46 :
47 0 : int32_t HcommSetNotifyWaitTimeOut(float timeOut)
48 : {
49 0 : if (std::isnan(timeOut) || timeOut < 0.0f || timeOut > static_cast<float>(UINT32_MAX)) {
50 0 : HCCL_ERROR("[%s] in cpu timeOut[%f] is invalid.", __func__, timeOut);
51 0 : return HCCL_E_PARA;
52 : }
53 0 : uint32_t timeOutInt = static_cast<uint32_t>(timeOut);
54 0 : HCCL_INFO("[%s] START in cpu. timeOut[%u].", __func__, timeOutInt);
55 0 : return g_threadLaunchCtx.SetNotifyWaitTimeOut(timeOutInt);
56 : }
57 :
58 0 : int32_t HcommThreadResAcquireTimeOut(float timeOut)
59 : {
60 0 : HCCL_ERROR("[%s] timeOut[%f], not support in cpu.", __func__, timeOut);
61 0 : return HCCL_E_NOT_SUPPORT;
62 : }
63 :
64 1 : int32_t HcommLocalCopyOnThread(ThreadHandle thread, void *dst, const void *src, uint64_t len)
65 : {
66 1 : HCCL_INFO("[%s] START. thread[0x%llx], dst[0x%llx], src[0x%llx], len[%llu].", __func__, thread, dst, src, len);
67 :
68 1 : CHK_PTR_NULL(dst);
69 1 : CHK_PTR_NULL(src);
70 1 : AddThreadWithTag(thread);
71 :
72 1 : Thread *const threadPtr = reinterpret_cast<Thread *>(thread);
73 1 : CHK_PTR_NULL(threadPtr);
74 :
75 1 : if (threadPtr->IsDeviceA5()) {
76 0 : CHK_RET(threadPtr->LocalCopy(dst, src, len));
77 : } else {
78 1 : HcclBuf srcBuf{const_cast<void *>(src), len, nullptr};
79 1 : HcclBuf dstBuf{dst, len, nullptr};
80 1 : Stream *stream = GetStream(thread);
81 1 : CHK_PTR_NULL(stream);
82 :
83 1 : HcclResult ret = HcclLocalCopy(stream, &dstBuf, &srcBuf);
84 1 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[%s] FAIL. thread[0x%llx], dst[0x%llx], src[0x%llx], len[%llu].",
85 : __func__, thread, dst, src, len), ret);
86 : }
87 1 : HCCL_INFO("[%s] SUCCESS.", __func__);
88 1 : return HCCL_SUCCESS;
89 : }
90 :
91 0 : int32_t HcommLocalReduceOnThread(ThreadHandle thread, void *dst, const void *src, uint64_t count,
92 : HcommDataType dataType, HcommReduceOp reduceOp)
93 : {
94 0 : HCCL_INFO("[%s] START. thread[0x%llx], dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d].",
95 : __func__, thread, dst, src, count, dataType, reduceOp);
96 :
97 0 : CHK_PTR_NULL(dst);
98 0 : CHK_PTR_NULL(src);
99 0 : CHK_PRT_RET((IsSupportReduce(dataType, reduceOp) == false), HCCL_ERROR("[HcommLocalReduceOnThread]Not support reduce, "
100 : "dst[%p], src[%p], count[%llu], dataType[%d], reduceOp[%d]", dst, src, count, dataType, reduceOp), HCCL_E_PARA);
101 0 : AddThreadWithTag(thread);
102 :
103 0 : Thread *const threadPtr = reinterpret_cast<Thread *>(thread);
104 0 : CHK_PTR_NULL(threadPtr);
105 :
106 0 : uint64_t len = count * SIZE_TABLE[dataType];
107 :
108 0 : if (threadPtr->IsDeviceA5()) {
109 0 : CHK_RET(threadPtr->LocalReduce(dst, src, len, dataType, reduceOp));
110 : } else {
111 0 : HcclBuf srcBuf{const_cast<void *>(src), len, nullptr};
112 0 : HcclBuf dstBuf{dst, len, nullptr};
113 0 : HcclReduceInfo reduceInfo{static_cast<HcclDataType>(dataType), static_cast<HcclReduceOp>(reduceOp)};
114 0 : Stream *stream = GetStream(thread);
115 0 : CHK_PTR_NULL(stream);
116 :
117 0 : HcclResult ret = HcclLocalCopyReduce(stream, &dstBuf, &srcBuf, reduceInfo);
118 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
119 : HCCL_ERROR("[%s] FAIL. thread[0x%llx], dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d].",
120 : __func__, thread, dst, src, count, dataType, reduceOp), ret);
121 : }
122 0 : HCCL_INFO("[%s] SUCCESS.", __func__);
123 0 : return HCCL_SUCCESS;
124 : }
125 :
126 0 : int32_t HcommThreadNotifyRecordOnThread(ThreadHandle thread, ThreadHandle dstThread, uint32_t dstNotifyIdx)
127 : {
128 0 : HCCL_INFO("[%s] START. thread[0x%llx], dstThread[0x%llx], dstNotifyIdx[%u].", __func__, thread, dstThread, dstNotifyIdx);
129 :
130 0 : AddThreadWithTag(thread);
131 :
132 0 : Thread *const threadPtr = reinterpret_cast<Thread *>(thread);
133 0 : CHK_PTR_NULL(threadPtr);
134 :
135 0 : if (threadPtr->IsDeviceA5()) {
136 0 : HcclResult ret = threadPtr->LocalNotifyRecord(dstThread, dstNotifyIdx);
137 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[%s] FAIL. thread[0x%llx], dstThread[0x%llx], notifyIdx[%u].",
138 : __func__, thread, dstThread, dstNotifyIdx), ret);
139 : } else {
140 0 : Stream *stream = GetStream(thread);
141 0 : CHK_PTR_NULL(stream);
142 :
143 0 : LocalNotify *notify = GetNotify(dstThread, dstNotifyIdx);
144 0 : CHK_PTR_NULL(notify);
145 :
146 0 : HcclResult ret = HcclLocalNotifyRecord(stream, notify);
147 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[%s] FAIL. thread[0x%llx], dstThread[0x%llx], notifyIdx[%u].",
148 : __func__, thread, dstThread, dstNotifyIdx), ret);
149 : }
150 :
151 0 : HCCL_INFO("[%s] SUCCESS.", __func__);
152 0 : return HCCL_SUCCESS;
153 : }
154 :
155 0 : int32_t HcommThreadNotifyWaitOnThread(ThreadHandle thread, uint32_t notifyIdx, uint32_t timeOut)
156 : {
157 0 : HCCL_INFO("[%s] START. thread[0x%llx], notifyIdx[%u], timeOut[%u].", __func__, thread, notifyIdx, timeOut);
158 :
159 0 : AddThreadWithTag(thread);
160 :
161 0 : Thread *const threadPtr = reinterpret_cast<Thread *>(thread);
162 0 : CHK_PTR_NULL(threadPtr);
163 :
164 0 : if (threadPtr->IsDeviceA5()) {
165 0 : HcclResult ret = threadPtr->LocalNotifyWait(notifyIdx, timeOut);
166 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[%s] FAIL. thread[0x%llx], notifyIdx[%u], timeOut[%u].",
167 : __func__, thread, notifyIdx, timeOut), ret);
168 : } else {
169 0 : Stream *stream = GetStream(thread);
170 0 : CHK_PTR_NULL(stream);
171 0 : LocalNotify *notify = GetNotify(thread, notifyIdx);
172 0 : CHK_PTR_NULL(notify);
173 :
174 0 : HcclResult ret = HcclLocalNotifyWait(stream, notify, timeOut);
175 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[%s] FAIL. thread[0x%llx], notifyIdx[%u], timeOut[%u].",
176 : __func__, thread, notifyIdx, timeOut), ret);
177 : }
178 0 : HCCL_INFO("[%s] SUCCESS.", __func__);
179 0 : return HCCL_SUCCESS;
180 : }
181 :
182 0 : int32_t HcommAclrtNotifyRecordOnThread(ThreadHandle thread, uint64_t dstNotifyId)
183 : {
184 0 : HCCL_INFO("[%s] START. thread[0x%llx], dstNotifyId[%u].", __func__, thread, dstNotifyId);
185 :
186 0 : AddThreadWithTag(thread);
187 :
188 0 : Thread *const threadPtr = reinterpret_cast<Thread *>(thread);
189 0 : CHK_PTR_NULL(threadPtr);
190 :
191 0 : Stream *stream = GetStream(thread);
192 0 : CHK_PTR_NULL(stream);
193 :
194 0 : HcclResult ret = HcclLocalBareNotifyRecord(stream, dstNotifyId);
195 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[%s] FAIL. thread[0x%llx], dstNotifyId[%u].", __func__, thread, dstNotifyId), ret);
196 0 : HCCL_INFO("[%s] SUCCESS.", __func__);
197 0 : return HCCL_SUCCESS;
198 : }
199 :
200 0 : int32_t HcommAclrtNotifyWaitOnThread(ThreadHandle thread, uint64_t notifyId, uint32_t timeOut)
201 : {
202 0 : HCCL_INFO("[%s] START. thread[0x%llx], notifyId[%llu], timeOut[%u].", __func__, thread, notifyId, timeOut);
203 :
204 0 : AddThreadWithTag(thread);
205 :
206 0 : Thread *const threadPtr = reinterpret_cast<Thread *>(thread);
207 0 : CHK_PTR_NULL(threadPtr);
208 :
209 0 : Stream *stream = GetStream(thread);
210 0 : CHK_PTR_NULL(stream);
211 :
212 0 : HcclResult ret = HcclLocalBareNotifyWait(stream, notifyId, timeOut);
213 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[%s] FAIL. thread[0x%llx], notifyId[%llu], timeOut[%u].", __func__, thread, notifyId, timeOut), ret);
214 0 : HCCL_INFO("[%s] SUCCESS.", __func__);
215 0 : return HCCL_SUCCESS;
216 : }
217 :
218 0 : HcclResult CommTaskPrepare(char *key, uint32_t keyLen) // host ffts+使用
219 : {
220 0 : std::string keyStr = "temp_key";
221 0 : if (key != nullptr && keyLen != 0) {
222 0 : keyStr = std::string(key, keyLen);
223 0 : HCCL_DEBUG("[CommTaskPrepare]key[%s], keyLen[%u]", key, keyLen);
224 : } else {
225 0 : HCCL_DEBUG("[CommTaskPrepare]disable cache, key[0x%llx], keyLen[%u]", key, keyLen);
226 : }
227 :
228 0 : return HcclTaskPrepare(const_cast<char_t*>(keyStr.c_str()), keyStr.length());
229 0 : }
230 :
231 0 : HcclResult CommTaskLaunch(ThreadHandle *threads, uint32_t threadNum) // host ffts+或aicpu stars使用"
232 : {
233 0 : CHK_PTR_NULL(threads);
234 0 : CHK_PRT_RET(threadNum < 1, HCCL_ERROR("[CommTaskLaunch]threadNum is less than 1"), HCCL_E_PARA);
235 :
236 0 : Thread *threadPtr = reinterpret_cast<Thread *>(threads[0]);
237 0 : CHK_PTR_NULL(threadPtr);
238 :
239 0 : std::vector<hccl::Stream> streams;
240 0 : for (uint32_t i = 0; i < threadNum; i++) {
241 0 : hccl::Stream *stream = GetStream(threads[i]);
242 0 : CHK_PTR_NULL(stream);
243 0 : streams.push_back(*stream);
244 : }
245 :
246 0 : return HcclTaskLaunch(streams.data(), threadNum);
247 0 : }
248 :
249 0 : HcclResult DispatchAllStreams(ThreadHandle *threads, uint32_t threadNum)
250 : {
251 0 : CHK_PTR_NULL(threads);
252 0 : CHK_PRT_RET(threadNum < 1, HCCL_ERROR("[DispatchAllStreams]threadNum is less than 1"), HCCL_E_PARA);
253 :
254 0 : HCCL_WARNING("[DispatchAllStreams] DispatchAllStreams is only supported on A5 device, skip");
255 0 : return HCCL_E_NOT_SUPPORT;
256 : }
257 :
258 2 : int32_t HcommWriteOnThread(ThreadHandle thread, ChannelHandle channel, void *dst, const void *src, uint64_t len)
259 : {
260 2 : HCCL_INFO("[%s] START. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu].",
261 : __func__, thread, channel, dst, src, len);
262 :
263 2 : CHK_PTR_NULL(dst);
264 2 : CHK_PTR_NULL(src);
265 : #ifdef ENABLE_EXPERIMENTAL
266 2 : bool handled = false;
267 2 : CHK_RET(static_cast<HcclResult>(hcomm::PluginChannelWrite(channel, thread, dst, src, len, handled)));
268 1 : if (handled) {
269 1 : return HCCL_SUCCESS;
270 : }
271 : #endif
272 :
273 0 : AddThreadWithTag(thread);
274 :
275 0 : Thread *const threadPtr = reinterpret_cast<Thread *>(thread);
276 0 : CHK_PTR_NULL(threadPtr);
277 :
278 0 : HcclBuf locBuf{const_cast<void *>(src), len, nullptr};
279 0 : HcclBuf rmtBuf{dst, len, nullptr};
280 :
281 0 : Stream *stream = GetStream(thread);
282 0 : CHK_PTR_NULL(stream);
283 :
284 0 : HcclResult ret = HcclRemoteWrite(stream, reinterpret_cast<void *>(channel), &rmtBuf, &locBuf);
285 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
286 : HCCL_ERROR("[%s] FAIL. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu].",
287 : __func__, thread, channel, dst, src, len), ret);
288 0 : HCCL_INFO("[%s] SUCCESS.", __func__);
289 0 : return HCCL_SUCCESS;
290 : }
291 :
292 1 : int32_t HcommBatchTransferOnThread(ThreadHandle thread, ChannelHandle channel,
293 : const HcommBatchTransferDesc *transferDescs, uint32_t transferDescNum)
294 : {
295 : #ifdef ENABLE_EXPERIMENTAL
296 1 : bool handled = false;
297 1 : CHK_RET(static_cast<HcclResult>(hcomm::PluginChannelBatchTransfer(channel, thread, transferDescs, transferDescNum,
298 : handled)));
299 0 : if (handled) {
300 0 : return HCCL_SUCCESS;
301 : }
302 : #endif
303 :
304 0 : HCCL_ERROR(" [HcommBatchTransferOnThread] not support in cpu");
305 0 : return HCCL_E_NOT_SUPPORT;
306 : }
307 :
308 :
309 1 : int32_t HcommWriteReduceOnThread(ThreadHandle thread, ChannelHandle channel, void *dst, const void *src,
310 : uint64_t count, HcommDataType dataType, HcommReduceOp reduceOp)
311 : {
312 1 : HCCL_INFO("[%s] START. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d].",
313 : __func__, thread, channel, dst, src, count, dataType, reduceOp);
314 :
315 1 : CHK_PTR_NULL(dst);
316 1 : CHK_PTR_NULL(src);
317 : #ifdef ENABLE_EXPERIMENTAL
318 1 : bool handled = false;
319 1 : CHK_RET(static_cast<HcclResult>(hcomm::PluginChannelWriteReduce(channel, thread, dst, src, count, dataType,
320 : reduceOp, handled)));
321 0 : if (handled) {
322 0 : return HCCL_SUCCESS;
323 : }
324 : #endif
325 :
326 0 : CHK_PRT_RET((IsSupportReduce(dataType, reduceOp) == false), HCCL_ERROR("[HcommWriteReduceOnThread]Not support reduce, "
327 : "dst[%p], src[%p], count[%llu], dataType[%d], reduceOp[%d]", dst, src, count, dataType, reduceOp), HCCL_E_PARA);
328 0 : AddThreadWithTag(thread);
329 :
330 0 : Thread *const threadPtr = reinterpret_cast<Thread *>(thread);
331 0 : CHK_PTR_NULL(threadPtr);
332 :
333 0 : uint64_t len = count * SIZE_TABLE[dataType];
334 :
335 0 : HcclBuf locBuf{const_cast<void *>(src), len, nullptr};
336 0 : HcclBuf rmtBuf{dst, len, nullptr};
337 0 : HcclReduceInfo reduceInfo{static_cast<HcclDataType>(dataType), static_cast<HcclReduceOp>(reduceOp)};
338 :
339 0 : Stream *stream = GetStream(thread);
340 0 : CHK_PTR_NULL(stream);
341 :
342 0 : HcclResult ret = HcclRemoteWriteReduce(stream, reinterpret_cast<void *>(channel), &rmtBuf, &locBuf, reduceInfo);
343 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
344 : HCCL_ERROR("[%s] FAIL. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d].",
345 : __func__, thread, channel, dst, src, count, dataType, reduceOp), ret);
346 0 : HCCL_INFO("[%s] SUCCESS.", __func__);
347 0 : return HCCL_SUCCESS;
348 : }
349 :
350 0 : HcclResult CommWriteReduceWithNotify(ThreadHandle thread, ChannelHandle channel, void *dst, const void *src,
351 : uint64_t count, HcommDataType dataType, HcommReduceOp reduceOp, uint32_t remoteNotifyIdx)
352 : {
353 0 : CHK_PTR_NULL(src);
354 0 : CHK_PTR_NULL(dst);
355 : #ifdef ENABLE_EXPERIMENTAL
356 0 : bool handled = false;
357 0 : CHK_RET(static_cast<HcclResult>(hcomm::PluginChannelWriteReduceWithNotify(channel, thread, dst, src, count,
358 : dataType, reduceOp, remoteNotifyIdx, handled)));
359 0 : if (handled) {
360 0 : return HCCL_SUCCESS;
361 : }
362 : #endif
363 :
364 0 : CHK_PRT_RET((IsSupportReduce(dataType, reduceOp) == false), HCCL_ERROR("[CommWriteReduceWithNotify]Not support reduce, "
365 : "dst[%p], src[%p], count[%llu], dataType[%d], reduceOp[%d]", dst, src, count, dataType, reduceOp), HCCL_E_PARA);
366 0 : AddThreadWithTag(thread);
367 0 : HcclBuf locBuf{const_cast<void*>(src), count * SIZE_TABLE[dataType], nullptr};
368 0 : HcclBuf rmtBuf{dst, count * SIZE_TABLE[dataType], nullptr};
369 0 : HcclReduceInfo reduceInfo{static_cast<HcclDataType>(dataType), static_cast<HcclReduceOp>(reduceOp)};
370 :
371 0 : Stream *stream = GetStream(thread);
372 0 : CHK_PTR_NULL(stream);
373 :
374 0 : return HcclRemoteWriteReduceWithNotify(stream, reinterpret_cast<void*>(channel), &rmtBuf, &locBuf, reduceInfo,
375 0 : remoteNotifyIdx);
376 : }
377 :
378 1 : int32_t HcommWriteWithNotifyOnThread(ThreadHandle thread, ChannelHandle channel, void *dst, const void *src,
379 : uint64_t len, uint32_t remoteNotifyIdx)
380 : {
381 1 : HCCL_INFO("[%s] START. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu], remoteNotifyIdx[%u].",
382 : __func__, thread, channel, dst, src, len, remoteNotifyIdx);
383 :
384 1 : CHK_PTR_NULL(src);
385 1 : CHK_PTR_NULL(dst);
386 : #ifdef ENABLE_EXPERIMENTAL
387 1 : bool handled = false;
388 1 : CHK_RET(static_cast<HcclResult>(hcomm::PluginChannelWriteWithNotify(channel, thread, dst, src, len,
389 : remoteNotifyIdx, handled)));
390 1 : if (handled) {
391 1 : return HCCL_SUCCESS;
392 : }
393 : #endif
394 :
395 0 : AddThreadWithTag(thread);
396 :
397 0 : Thread *const threadPtr = reinterpret_cast<Thread *>(thread);
398 0 : CHK_PTR_NULL(threadPtr);
399 :
400 0 : HcclBuf locBuf{const_cast<void *>(src), len, nullptr};
401 0 : HcclBuf rmtBuf{dst, len, nullptr};
402 :
403 0 : Stream *stream = GetStream(thread);
404 0 : CHK_PTR_NULL(stream);
405 :
406 0 : HcclResult ret = HcclRemoteWriteWithNotify(stream, reinterpret_cast<void *>(channel), &rmtBuf, &locBuf, remoteNotifyIdx);
407 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
408 : HCCL_ERROR("[%s] FAIL. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu], remoteNotifyIdx[%u].",
409 : __func__, thread, channel, dst, src, len, remoteNotifyIdx), ret);
410 0 : HCCL_INFO("[%s] SUCCESS.", __func__);
411 0 : return HCCL_SUCCESS;
412 : }
413 :
414 1 : int32_t HcommWriteReduceWithNotifyOnThread(ThreadHandle thread, ChannelHandle channel, void *dst,
415 : const void *src, uint64_t count, HcommDataType dataType, HcommReduceOp reduceOp, uint32_t remoteNotifyIdx)
416 : {
417 1 : HCCL_INFO("[%s] START. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d], remoteNotifyIdx[%u].",
418 : __func__, thread, channel, dst, src, count, dataType, reduceOp, remoteNotifyIdx);
419 :
420 1 : CHK_PTR_NULL(dst);
421 1 : CHK_PTR_NULL(src);
422 : #ifdef ENABLE_EXPERIMENTAL
423 1 : bool handled = false;
424 1 : CHK_RET(static_cast<HcclResult>(hcomm::PluginChannelWriteReduceWithNotify(channel, thread, dst, src, count,
425 : dataType, reduceOp, remoteNotifyIdx, handled)));
426 0 : if (handled) {
427 0 : return HCCL_SUCCESS;
428 : }
429 : #endif
430 :
431 0 : AddThreadWithTag(thread);
432 :
433 0 : Thread *const threadPtr = reinterpret_cast<Thread *>(thread);
434 0 : CHK_PTR_NULL(threadPtr);
435 :
436 0 : uint64_t len = count * SIZE_TABLE[dataType];
437 :
438 0 : HcclResult ret = HCCL_SUCCESS;
439 :
440 0 : ret = HCCL_E_NOT_SUPPORT;
441 : (void)len;
442 :
443 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
444 : HCCL_ERROR("[%s] FAIL. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d], remoteNotifyIdx[%u].",
445 : __func__, thread, channel, dst, src, count, dataType, reduceOp, remoteNotifyIdx), ret);
446 0 : HCCL_INFO("[%s] SUCCESS.", __func__);
447 0 : return HCCL_SUCCESS;
448 : }
449 :
450 1 : int32_t HcommReadOnThread(ThreadHandle thread, ChannelHandle channel, void *dst, const void *src, uint64_t len)
451 : {
452 1 : HCCL_INFO("[%s] START. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu].",
453 : __func__, thread, channel, dst, src, len);
454 :
455 1 : CHK_PTR_NULL(dst);
456 1 : CHK_PTR_NULL(src);
457 : #ifdef ENABLE_EXPERIMENTAL
458 1 : bool handled = false;
459 1 : CHK_RET(static_cast<HcclResult>(hcomm::PluginChannelRead(channel, thread, dst, src, len, handled)));
460 1 : if (handled) {
461 1 : return HCCL_SUCCESS;
462 : }
463 : #endif
464 :
465 0 : AddThreadWithTag(thread);
466 :
467 0 : Thread *const threadPtr = reinterpret_cast<Thread *>(thread);
468 0 : CHK_PTR_NULL(threadPtr);
469 :
470 0 : HcclBuf locBuf{dst, len, nullptr};
471 0 : HcclBuf rmtBuf{const_cast<void *>(src), len, nullptr};
472 :
473 0 : Stream *stream = GetStream(thread);
474 0 : CHK_PTR_NULL(stream);
475 :
476 0 : HcclResult ret = HcclRemoteRead(stream, reinterpret_cast<void *>(channel), &locBuf, &rmtBuf);
477 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
478 : HCCL_ERROR("[%s] FAIL. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu].",
479 : __func__, thread, channel, dst, src, len), ret);
480 0 : HCCL_INFO("[%s] SUCCESS.", __func__);
481 0 : return HCCL_SUCCESS;
482 : }
483 :
484 1 : int32_t HcommReadReduceOnThread(ThreadHandle thread, ChannelHandle channel, void *dst, const void *src,
485 : uint64_t count, HcommDataType dataType, HcommReduceOp reduceOp)
486 : {
487 1 : HCCL_INFO("[%s] START. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d].",
488 : __func__, thread, channel, dst, src, count, dataType, reduceOp);
489 :
490 1 : CHK_PTR_NULL(dst);
491 1 : CHK_PTR_NULL(src);
492 : #ifdef ENABLE_EXPERIMENTAL
493 1 : bool handled = false;
494 1 : CHK_RET(static_cast<HcclResult>(hcomm::PluginChannelReadReduce(channel, thread, dst, src, count, dataType,
495 : reduceOp, handled)));
496 0 : if (handled) {
497 0 : return HCCL_SUCCESS;
498 : }
499 : #endif
500 :
501 0 : CHK_PRT_RET((IsSupportReduce(dataType, reduceOp) == false), HCCL_ERROR("[HcommReadReduceOnThread]Not support reduce, "
502 : "dst[%p], src[%p], count[%llu], dataType[%d], reduceOp[%d]", dst, src, count, dataType, reduceOp), HCCL_E_PARA);
503 0 : AddThreadWithTag(thread);
504 :
505 0 : Thread *const threadPtr = reinterpret_cast<Thread *>(thread);
506 0 : CHK_PTR_NULL(threadPtr);
507 :
508 0 : uint64_t len = count * SIZE_TABLE[dataType];
509 :
510 0 : HcclBuf locBuf{dst, len, nullptr};
511 0 : HcclBuf rmtBuf{const_cast<void *>(src), len, nullptr};
512 0 : HcclReduceInfo reduceInfo{static_cast<HcclDataType>(dataType), static_cast<HcclReduceOp>(reduceOp)};
513 :
514 0 : Stream *stream = GetStream(thread);
515 0 : CHK_PTR_NULL(stream);
516 :
517 0 : HcclResult ret = HcclRemoteReadReduce(stream, reinterpret_cast<void *>(channel), &locBuf, &rmtBuf, reduceInfo);
518 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
519 : HCCL_ERROR("[%s] FAIL. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d].",
520 : __func__, thread, channel, dst, src, count, dataType, reduceOp), ret);
521 0 : HCCL_INFO("[%s] SUCCESS.", __func__);
522 0 : return HCCL_SUCCESS;
523 : }
524 :
525 8 : int32_t HcommWriteNbiOnThread(ThreadHandle thread, ChannelHandle channel, void *dst, const void *src, uint64_t len)
526 : {
527 8 : HCCL_INFO("[%s] START. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu].",
528 : __func__, thread, channel, dst, src, len);
529 :
530 : (void)thread;
531 8 : CHK_PTR_NULL(src);
532 7 : CHK_PTR_NULL(dst);
533 : #ifdef ENABLE_EXPERIMENTAL
534 6 : bool handled = false;
535 6 : CHK_RET(static_cast<HcclResult>(hcomm::PluginChannelWriteNbi(channel, thread, dst, src, len, handled)));
536 6 : if (handled) {
537 1 : return HCCL_SUCCESS;
538 : }
539 : #endif
540 :
541 5 : HcclResult ret = HCCL_SUCCESS;
542 : DevType devType;
543 5 : CHK_RET(hrtGetDeviceType(devType));
544 5 : if (devType == DevType::DEV_TYPE_950 || devType == DevType::DEV_TYPE_960) {
545 4 : auto *const channelPtr = reinterpret_cast<hcomm::Channel *>(channel);
546 4 : CHK_PTR_NULL(channelPtr);
547 3 : ret = channelPtr->Write(dst, src, len);
548 3 : } else {
549 1 : ret = HCCL_E_NOT_SUPPORT;
550 : }
551 4 : CHK_PRT_RET(ret != HCCL_SUCCESS,
552 : HCCL_ERROR("[%s] FAIL. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu].",
553 : __func__, thread, channel, dst, src, len), ret);
554 2 : HCCL_INFO("[%s] SUCCESS.", __func__);
555 2 : return HCCL_SUCCESS;
556 : }
557 :
558 2 : int32_t HcommWriteNbi(ChannelHandle channel, void *dst, const void *src, uint64_t len)
559 : {
560 : #ifdef ENABLE_EXPERIMENTAL
561 2 : bool handled = false;
562 2 : CHK_RET(static_cast<HcclResult>(hcomm::PluginChannelWriteNbi(channel, 0, dst, src, len, handled)));
563 1 : if (handled) {
564 1 : return HCCL_SUCCESS;
565 : }
566 : #endif
567 0 : return HcommWriteNbiOnThread(0, channel, dst, src, len);
568 : }
569 :
570 8 : int32_t HcommWriteWithNotifyNbiOnThread(ThreadHandle thread, ChannelHandle channel, void *dst, const void *src,
571 : uint64_t len, uint32_t remoteNotifyIdx)
572 : {
573 8 : HCCL_INFO("[%s] START. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu], remoteNotifyIdx[%u].",
574 : __func__, thread, channel, dst, src, len, remoteNotifyIdx);
575 :
576 : (void)thread;
577 8 : CHK_PTR_NULL(src);
578 7 : CHK_PTR_NULL(dst);
579 : #ifdef ENABLE_EXPERIMENTAL
580 6 : bool handled = false;
581 6 : CHK_RET(static_cast<HcclResult>(hcomm::PluginChannelWriteWithNotifyNbi(channel, thread, dst, src, len,
582 : remoteNotifyIdx, handled)));
583 6 : CHK_PRT_RET(handled, HCCL_INFO("[%s] SUCCESS.", __func__), HCCL_SUCCESS);
584 : #endif
585 :
586 5 : HcclResult ret = HCCL_SUCCESS;
587 : DevType devType;
588 5 : CHK_RET(hrtGetDeviceType(devType));
589 5 : if (devType == DevType::DEV_TYPE_950 || devType == DevType::DEV_TYPE_960 || thread == 0) {
590 4 : auto *const channelPtr = reinterpret_cast<hcomm::Channel *>(channel);
591 4 : CHK_PTR_NULL(channelPtr);
592 3 : ret = channelPtr->WriteWithNotify(dst, src, len, remoteNotifyIdx);
593 3 : } else {
594 1 : ret = HCCL_E_NOT_SUPPORT;
595 : }
596 4 : CHK_PRT_RET(ret != HCCL_SUCCESS,
597 : HCCL_ERROR("[%s] FAIL. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu], remoteNotifyIdx[%u].",
598 : __func__, thread, channel, dst, src, len, remoteNotifyIdx), ret);
599 2 : HCCL_INFO("[%s] SUCCESS.", __func__);
600 2 : return HCCL_SUCCESS;
601 : }
602 :
603 1 : int32_t HcommWriteWithNotifyNbi(ChannelHandle channel, void *dst, const void *src,
604 : uint64_t len, uint32_t remoteNotifyIdx)
605 : {
606 : #ifdef ENABLE_EXPERIMENTAL
607 1 : bool handled = false;
608 1 : CHK_RET(static_cast<HcclResult>(hcomm::PluginChannelWriteWithNotifyNbi(channel, 0, dst, src, len, remoteNotifyIdx,
609 : handled)));
610 1 : if (handled) {
611 1 : return HCCL_SUCCESS;
612 : }
613 : #endif
614 0 : return HcommWriteWithNotifyNbiOnThread(0, channel, dst, src, len, remoteNotifyIdx);
615 : }
616 :
617 8 : int32_t HcommReadNbiOnThread(ThreadHandle thread, ChannelHandle channel, void *dst, const void *src, uint64_t len)
618 : {
619 8 : HCCL_INFO("[%s] START. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu].",
620 : __func__, thread, channel, dst, src, len);
621 :
622 : (void)thread;
623 8 : CHK_PTR_NULL(src);
624 7 : CHK_PTR_NULL(dst);
625 : #ifdef ENABLE_EXPERIMENTAL
626 6 : bool handled = false;
627 6 : CHK_RET(static_cast<HcclResult>(hcomm::PluginChannelReadNbi(channel, thread, dst, src, len, handled)));
628 6 : if (handled) {
629 1 : return HCCL_SUCCESS;
630 : }
631 : #endif
632 :
633 5 : HcclResult ret = HCCL_SUCCESS;
634 : DevType devType;
635 5 : CHK_RET(hrtGetDeviceType(devType));
636 5 : if (devType == DevType::DEV_TYPE_950 || devType == DevType::DEV_TYPE_960) {
637 4 : auto *const channelPtr = reinterpret_cast<hcomm::Channel *>(channel);
638 4 : CHK_PTR_NULL(channelPtr);
639 3 : ret = channelPtr->Read(dst, src, len);
640 3 : } else {
641 1 : ret = HCCL_E_NOT_SUPPORT;
642 : }
643 4 : CHK_PRT_RET(ret != HCCL_SUCCESS,
644 : HCCL_ERROR("[%s] FAIL. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu].",
645 : __func__, thread, channel, dst, src, len), ret);
646 2 : HCCL_INFO("[%s] SUCCESS.", __func__);
647 2 : return HCCL_SUCCESS;
648 : }
649 :
650 1 : int32_t HcommReadNbi(ChannelHandle channel, void *dst, const void *src, uint64_t len)
651 : {
652 : #ifdef ENABLE_EXPERIMENTAL
653 1 : bool handled = false;
654 1 : CHK_RET(static_cast<HcclResult>(hcomm::PluginChannelReadNbi(channel, 0, dst, src, len, handled)));
655 1 : if (handled) {
656 1 : return HCCL_SUCCESS;
657 : }
658 : #endif
659 0 : return HcommReadNbiOnThread(0, channel, dst, src, len);
660 : }
661 :
662 9 : int32_t HcommChannelNotifyRecordOnThread(ThreadHandle thread, ChannelHandle channel, uint32_t remoteNotifyIdx)
663 : {
664 9 : HCCL_INFO("[%s] START. thread[0x%llx], channel[0x%llx], remoteNotifyIdx[%u].", __func__, thread, channel, remoteNotifyIdx);
665 :
666 : #ifdef ENABLE_EXPERIMENTAL
667 9 : bool handled = false;
668 9 : CHK_RET(static_cast<HcclResult>(hcomm::PluginChannelNotifyRecord(channel, thread, remoteNotifyIdx, handled)));
669 9 : if (handled) {
670 1 : return HCCL_SUCCESS;
671 : }
672 : #endif
673 :
674 8 : HcclResult ret = HCCL_SUCCESS;
675 : DevType devType;
676 8 : CHK_RET(hrtGetDeviceType(devType));
677 8 : if (devType == DevType::DEV_TYPE_950 || devType == DevType::DEV_TYPE_960 ||
678 1 : (thread == 0 && devType == DevType::DEV_TYPE_910B)) {
679 4 : auto *const channelPtr = reinterpret_cast<hcomm::Channel *>(channel);
680 4 : CHK_PTR_NULL(channelPtr);
681 3 : ret = channelPtr->NotifyRecord(remoteNotifyIdx);
682 3 : } else { // Non-950 devices use thread-based notify.
683 4 : AddThreadWithTag(thread);
684 :
685 4 : Thread *threadPtr = reinterpret_cast<Thread *>(thread);
686 4 : CHK_PTR_NULL(threadPtr);
687 :
688 3 : Stream *stream = GetStream(thread);
689 3 : CHK_PTR_NULL(stream);
690 :
691 2 : ret = HcclRemoteNotifyRecord(stream, reinterpret_cast<void *>(channel), remoteNotifyIdx);
692 : }
693 5 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[%s] FAIL. thread[0x%llx], channel[0x%llx], remoteNotifyIdx[%u].", __func__, thread, channel, remoteNotifyIdx), ret);
694 3 : HCCL_INFO("[%s] SUCCESS.", __func__);
695 3 : return HCCL_SUCCESS;
696 : }
697 :
698 2 : int32_t HcommChannelNotifyRecord(ChannelHandle channel, uint32_t remoteNotifyIdx)
699 : {
700 : #ifdef ENABLE_EXPERIMENTAL
701 2 : bool handled = false;
702 2 : CHK_RET(static_cast<HcclResult>(hcomm::PluginChannelNotifyRecord(channel, 0, remoteNotifyIdx, handled)));
703 1 : if (handled) {
704 1 : return HCCL_SUCCESS;
705 : }
706 : #endif
707 :
708 : DevType devType;
709 0 : CHK_RET(hrtGetDeviceType(devType));
710 0 : if (devType != DevType::DEV_TYPE_950 && devType != DevType::DEV_TYPE_960) {
711 0 : return HCCL_E_NOT_SUPPORT;
712 : }
713 0 : return HcommChannelNotifyRecordOnThread(0, channel, remoteNotifyIdx);
714 : }
715 :
716 9 : int32_t HcommChannelNotifyWaitOnThread(ThreadHandle thread, ChannelHandle channel, uint32_t localNotifyIdx, uint32_t timeOut)
717 : {
718 9 : HCCL_INFO("[%s] START. thread[0x%llx], channel[0x%llx], localNotifyIdx[%u], timeOut[%u].", __func__, thread, channel, localNotifyIdx, timeOut);
719 :
720 : #ifdef ENABLE_EXPERIMENTAL
721 9 : bool handled = false;
722 9 : CHK_RET(static_cast<HcclResult>(hcomm::PluginChannelNotifyWait(channel, thread, localNotifyIdx, timeOut,
723 : handled)));
724 9 : if (handled) {
725 1 : return HCCL_SUCCESS;
726 : }
727 : #endif
728 :
729 8 : HcclResult ret = HCCL_SUCCESS;
730 : DevType devType;
731 8 : CHK_RET(hrtGetDeviceType(devType));
732 8 : if (devType == DevType::DEV_TYPE_950 || devType == DevType::DEV_TYPE_960 ||
733 1 : (thread == 0 && devType == DevType::DEV_TYPE_910B)) {
734 4 : auto *const channelPtr = reinterpret_cast<hcomm::Channel *>(channel);
735 4 : CHK_PTR_NULL(channelPtr);
736 3 : ret = channelPtr->NotifyWait(localNotifyIdx, timeOut);
737 3 : } else { // Non-950 devices use thread-based notify.
738 4 : AddThreadWithTag(thread);
739 :
740 4 : Thread *threadPtr = reinterpret_cast<Thread *>(thread);
741 4 : CHK_PTR_NULL(threadPtr);
742 :
743 3 : Stream *stream = GetStream(thread);
744 3 : CHK_PTR_NULL(stream);
745 :
746 2 : ret = HcclRemoteNotifyWait(stream, reinterpret_cast<void *>(channel), localNotifyIdx, timeOut);
747 : }
748 5 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[%s] FAIL. thread[0x%llx], channel[0x%llx], localNotifyIdx[%u], timeOut[%u]s.", __func__, thread, channel, localNotifyIdx, timeOut), ret);
749 3 : HCCL_INFO("[%s] SUCCESS.", __func__);
750 3 : return HCCL_SUCCESS;
751 : }
752 :
753 1 : int32_t HcommChannelNotifyWait(ChannelHandle channel, uint32_t localNotifyIdx, uint32_t timeOut)
754 : {
755 : #ifdef ENABLE_EXPERIMENTAL
756 1 : bool handled = false;
757 1 : CHK_RET(static_cast<HcclResult>(hcomm::PluginChannelNotifyWait(channel, 0, localNotifyIdx, timeOut, handled)));
758 1 : if (handled) {
759 1 : return HCCL_SUCCESS;
760 : }
761 : #endif
762 :
763 : DevType devType;
764 0 : CHK_RET(hrtGetDeviceType(devType));
765 0 : if (devType != DevType::DEV_TYPE_950 && devType != DevType::DEV_TYPE_960) {
766 0 : return HCCL_E_NOT_SUPPORT;
767 : }
768 0 : return HcommChannelNotifyWaitOnThread(0, channel, localNotifyIdx, timeOut);
769 : }
770 :
771 1 : HcclResult CommFence(ThreadHandle thread, ChannelHandle channel) // 控制前后的任务保序
772 : {
773 1 : HCCL_DEBUG("[CommFence] thread[0x%llx], channel[0x%llx].", thread, channel);
774 : #ifdef ENABLE_EXPERIMENTAL
775 1 : bool handled = false;
776 1 : CHK_RET(static_cast<HcclResult>(hcomm::PluginChannelFence(channel, thread, handled)));
777 1 : if (handled) {
778 1 : return HCCL_SUCCESS;
779 : }
780 : #endif
781 :
782 0 : Stream *stream = GetStream(thread);
783 0 : CHK_PTR_NULL(stream);
784 :
785 0 : return HcclRemoteFence(stream, reinterpret_cast<void *>(channel), false);
786 : }
787 :
788 0 : int32_t HcommSetLaunchMode(const char *launchTag, HcommLaunchMode mode)
789 : {
790 0 : HCCL_DEBUG("HcommSetLaunchMode launchTag[%s]", launchTag);
791 0 : return g_threadLaunchCtx.SetLaunchMode(launchTag, mode);
792 : }
793 :
794 0 : int32_t HcommBatchModeStart(const char *batchTag)
795 : {
796 0 : return HcommSetLaunchMode(batchTag, HCOMM_LAUNCH_MODE_BATCH);
797 : }
798 :
799 0 : int32_t HcommBatchModeEnd(const char *batchTag)
800 : {
801 0 : return HcommSetLaunchMode(batchTag, HCOMM_LAUNCH_MODE_EAGER);
802 : }
803 :
804 23 : int32_t HcommThreadRegisterDfx(ThreadHandle thread, std::function<HcclResult(u32, u32, const Hccl::TaskParam&, u64)> callback)
805 : {
806 23 : Thread *threadPtr = reinterpret_cast<Thread *>(thread);
807 23 : CHK_PTR_NULL(threadPtr);
808 23 : CHK_RET(threadPtr->SetAddTaskInfoCallback(callback));
809 23 : return HCCL_SUCCESS;
810 : }
811 :
812 2 : int32_t HcommThreadRegisterCheckExecStatus(ThreadHandle thread, std::function<HcclResult(bool)> callback)
813 : {
814 2 : Thread *threadPtr = reinterpret_cast<Thread *>(thread);
815 2 : CHK_PTR_NULL(threadPtr);
816 2 : CHK_RET(threadPtr->SetCheckExecStatusCallback(callback));
817 2 : return HCCL_SUCCESS;
818 : }
819 :
820 4 : int32_t HcommDpuChannelRegisterDfx(ChannelHandle channel, std::function<HcclResult(const Hccl::TaskParam&, u64)> callback) {
821 4 : auto *const hostCpuRoceChannelPtr = reinterpret_cast<hcomm::HostCpuRoceChannel *>(channel);
822 4 : CHK_PTR_NULL(hostCpuRoceChannelPtr);
823 3 : CHK_RET(hostCpuRoceChannelPtr->SetDfxCallback(callback));
824 2 : return HCCL_SUCCESS;
825 : }
826 :
827 0 : int32_t HcommAcquireComm(const char* commId)
828 : {
829 0 : CHK_PTR_NULL(commId);
830 0 : std::shared_ptr<hccl::hcclComm> hcclComm;
831 0 : HcclGetCommHandle(commId, hcclComm);
832 0 : CHK_PRT_RET(hcclComm == nullptr, HCCL_ERROR("%s hcclComm is null, commId[%s]", __func__, commId), HCCL_E_PTR);
833 0 : CHK_RET(hcclComm->SetCommDispatcherCtx());
834 0 : return HCCL_SUCCESS;
835 0 : }
836 :
837 0 : int32_t HcommReleaseComm(const char* commId)
838 : {
839 0 : CHK_PTR_NULL(commId);
840 0 : HCCL_INFO("%s not support, commId[%s], do nothing", __func__, commId);
841 0 : return HCCL_SUCCESS;
842 : }
843 :
844 3 : int32_t HcommFenceOnThread(ThreadHandle thread)
845 : {
846 3 : HCCL_INFO("[%s] START. thread[0x%llx].", __func__, thread);
847 : (void)thread;
848 3 : HcclResult ret = HcommFlushV2();
849 3 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[%s] FAIL. thread[0x%llx].", __func__, thread), ret);
850 2 : HCCL_INFO("[%s] SUCCESS.", __func__);
851 2 : return HCCL_SUCCESS;
852 : }
853 :
854 0 : int32_t HcommChannelDrainOnThread(ThreadHandle thread, ChannelHandle channel)
855 : {
856 0 : Thread *const threadPtr = reinterpret_cast<Thread *>(thread);
857 0 : CHK_PTR_NULL(threadPtr);
858 0 : Stream *stream = GetStream(thread);
859 0 : CHK_PTR_NULL(stream);
860 :
861 0 : HcclResult ret = HcclRemoteDrain(stream, reinterpret_cast<void *>(channel));
862 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
863 : HCCL_ERROR("[%s] FAIL. thread[0x%llx], channel[0x%llx].", __func__, thread, channel), ret);
864 :
865 0 : return HCCL_SUCCESS;
866 : }
867 :
868 : #ifdef __cplusplus
869 : extern "C" {
870 : #endif // __cplusplus
871 0 : int32_t HcommFlush()
872 : {
873 0 : return HcommFenceOnThread(0);
874 : }
875 : #ifdef __cplusplus
876 : }
877 : #endif // __cplusplus
878 :
879 6 : int32_t HcommChannelFenceOnThread(ThreadHandle thread, ChannelHandle channel)
880 : {
881 6 : HCCL_INFO("[%s] START. thread[0x%llx], channel[0x%llx].", __func__, thread, channel);
882 :
883 : (void)thread;
884 :
885 : #ifdef ENABLE_EXPERIMENTAL
886 6 : bool handled = false;
887 6 : CHK_RET(static_cast<HcclResult>(hcomm::PluginChannelFence(channel, thread, handled)));
888 6 : CHK_PRT_RET(handled, HCCL_INFO("[%s] SUCCESS.", __func__), HCCL_SUCCESS);
889 : #endif
890 :
891 5 : HcclResult ret = HCCL_SUCCESS;
892 : DevType devType;
893 5 : CHK_RET(hrtGetDeviceType(devType));
894 5 : if (devType == DevType::DEV_TYPE_950 || devType == DevType::DEV_TYPE_960 || thread == 0) {
895 4 : auto *const channelPtr = reinterpret_cast<hcomm::Channel *>(channel);
896 4 : CHK_PTR_NULL(channelPtr);
897 3 : ret = channelPtr->ChannelFence();
898 3 : } else {
899 1 : ret = HCCL_E_NOT_SUPPORT;
900 : }
901 4 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[%s] FAIL. thread[0x%llx], channel[0x%llx].", __func__, thread, channel), ret);
902 2 : HCCL_INFO("[%s] SUCCESS.", __func__);
903 2 : return HCCL_SUCCESS;
904 : }
905 :
906 2 : int32_t HcommChannelFence(ChannelHandle channel)
907 : {
908 : #ifdef ENABLE_EXPERIMENTAL
909 2 : bool handled = false;
910 2 : CHK_RET(static_cast<HcclResult>(hcomm::PluginChannelFence(channel, 0, handled)));
911 1 : if (handled) {
912 1 : return HCCL_SUCCESS;
913 : }
914 : #endif
915 0 : return HcommChannelFenceOnThread(0, channel);
916 : }
917 :
918 0 : HcclResult HcclDfxRegOpInfo(HcclComm comm, void* hcclDfxOpInfo) // 兼容性接口，后续删除
919 : {
920 0 : HCCL_WARNING("%s not support", __func__);
921 0 : return HCCL_SUCCESS;
922 : }
923 :
924 0 : HcclResult HcclDfxRegOpInfoByCommId(char* commId, void* hcclDfxOpInfo)
925 : {
926 : EXCEPTION_HANDLE_BEGIN
927 0 : HcclComm commHandle = nullptr;
928 0 : CHK_RET(HcomGetCommHandleByGroup(commId, &commHandle));
929 0 : hccl::hcclComm* hcclComm = static_cast<hccl::hcclComm*>(commHandle);
930 0 : CHK_PRT_RET(hcclComm == nullptr, HCCL_ERROR("%s hcclComm is null, commId[%s]", __func__, commId), HCCL_E_PTR);
931 0 : CHK_PRT_RET(hcclDfxOpInfo == nullptr, HCCL_ERROR("[%s] hcclDfxOpInfo is null", __func__), HCCL_E_PTR);
932 0 : HcclDfxOpInfo *dfxOpInfo = static_cast<HcclDfxOpInfo*>(hcclDfxOpInfo);
933 0 : CHK_PTR_NULL(dfxOpInfo);
934 : DevType devType;
935 0 : CHK_RET(hrtGetDeviceType(devType));
936 0 : if (!hcclComm->IsCommunicatorV2() && devType == DevType::DEV_TYPE_910B) {
937 0 : return HCCL_SUCCESS;
938 : }
939 0 : if (!hcclComm->IsCommunicatorV2()) {
940 0 : HCCL_ERROR("[%s]comm is NOT_SUPPORT", __func__);
941 0 : return HCCL_E_NOT_SUPPORT;
942 : }
943 0 : hccl::CollComm* collComm = hcclComm->GetCollComm();
944 0 : CHK_PTR_NULL(collComm);
945 :
946 0 : dfxOpInfo->beginTime = hrtMsprofSysCycleTime();
947 :
948 : //HcclDfxOpInfo转为DfxOpInfo
949 0 : auto dfxOpInfoOnce = ConvertToDfxOpInfo(*dfxOpInfo);
950 0 : CHK_SMART_PTR_NULL(dfxOpInfoOnce);
951 0 : dfxOpInfoOnce->comm_ = static_cast<void*>(collComm);
952 0 : dfxOpInfoOnce->isIndop_ = true;
953 0 : dfxOpInfoOnce->groupName_ = collComm->GetCommId();
954 0 : dfxOpInfoOnce->opIndex_ = collComm->UpdateIndex();
955 0 : dfxOpInfoOnce->rankSize_ = collComm->GetRankSize();
956 : //单算子模式，暂时覆盖opTag
957 0 : dfxOpInfoOnce->op_.opTag = collComm->GetCommId();
958 0 : dfxOpInfoOnce->op_.myRank = static_cast<Hccl::RankId>(collComm->GetMyRankId());
959 0 : dfxOpInfoOnce->engine = dfxOpInfo->engine;
960 0 : HcclCommDfx* hcclCommDfx = collComm->GetHcclCommDfx();
961 0 : CHK_PTR_NULL(hcclCommDfx);
962 0 : CHK_RET(hcclCommDfx->UpdateProfStat());
963 0 : CHK_RET(hcclCommDfx->SetCurrDfxOpInfo(dfxOpInfoOnce));
964 0 : bool isOpBase = dfxOpInfoOnce->op_.opMode == Hccl::OpMode::OPBASE;
965 0 : Hccl::ProfilingHandler::GetInstance().SetIsOpbase(isOpBase);
966 0 : HCCL_INFO("[%s] Register DfxOpInfo success, commId[%s], opMode[%d], isOpBase[%d].", __func__, collComm->GetCommId().c_str(), dfxOpInfoOnce->op_.opMode, isOpBase);
967 0 : HCCL_INFO("[%s]success, DfxOpInfo: %s", __func__, dfxOpInfoOnce->Describe().c_str());
968 0 : EXCEPTION_HANDLE_END
969 0 : return HCCL_SUCCESS;
970 : }
971 :
972 0 : HcclResult HcclProfilingReportOp(HcclComm comm, uint64_t beginTime)
973 : {
974 0 : HCCL_INFO("[%s] START, comm[%p].", __func__, comm);
975 0 : CHK_PRT_RET(comm == nullptr, HCCL_ERROR("[%s] comm is null", __func__), HCCL_E_PTR);
976 0 : auto* hcclComm = static_cast<hccl::hcclComm*>(comm);
977 0 : CHK_PTR_NULL(hcclComm);
978 : DevType devType;
979 0 : CHK_RET(hrtGetDeviceType(devType));
980 0 : if (devType == DevType::DEV_TYPE_910B && !hcclComm->IsCommunicatorV2()) {
981 0 : return HCCL_SUCCESS;
982 : }
983 0 : if (!hcclComm->IsCommunicatorV2()) {
984 0 : HCCL_ERROR("[%s] comm is NOT_SUPPORT", __func__);
985 0 : return HCCL_E_NOT_SUPPORT;
986 : }
987 0 : hccl::CollComm* collComm = hcclComm->GetCollComm();
988 0 : CHK_PTR_NULL(collComm);
989 0 : HcclCommDfx* hcclCommDfx = collComm->GetHcclCommDfx();
990 0 : CHK_PTR_NULL(hcclCommDfx);
991 0 : HCCL_INFO("[%s] Report All Tasks Info, comm[%p], hcclCommDfx[%p] GetMirrorTaskManager[%p].",
992 : __func__, comm, hcclCommDfx, hcclCommDfx->GetMirrorTaskManager());
993 0 : auto* mirrorTaskMgr = hcclCommDfx->GetMirrorTaskManager();
994 0 : CHK_PTR_NULL(mirrorTaskMgr);
995 0 : if (mirrorTaskMgr->GetCurrDfxOpInfo() == nullptr) {
996 0 : HCCL_INFO("[%s] commId[%s] currDfxOpInfo is null, skip report.", __func__, collComm->GetCommId().c_str());
997 0 : return HCCL_SUCCESS;
998 : }
999 : //单算子模式暂时默认true
1000 0 : bool isBaseOpMode = false;
1001 0 : CHK_RET(hcclCommDfx->IsOpBase(isBaseOpMode));
1002 0 : CHK_RET(hcclCommDfx->ReportAllTasks(!isBaseOpMode));
1003 0 : CHK_RET(hcclCommDfx->ReportOp(beginTime, !isBaseOpMode, isBaseOpMode));
1004 0 : HCCL_INFO("[%s] SUCCESS.", __func__);
1005 0 : return HCCL_SUCCESS;
1006 : }
1007 :
1008 0 : HcclResult HcclReportAicpuKernel(HcclComm comm, uint64_t beginTime, char* kernelName)
1009 : {
1010 0 : HCCL_INFO("[%s] START, comm[%p].", __func__, comm);
1011 0 : CHK_PRT_RET(comm == nullptr, HCCL_ERROR("[%s] comm is null", __func__), HCCL_E_PTR);
1012 0 : CHK_PRT_RET(kernelName == nullptr, HCCL_ERROR("[%s] kernelName is null", __func__), HCCL_E_PTR);
1013 : //填入remoteRankId
1014 0 : auto hcclComm = static_cast<hccl::hcclComm*>(comm);
1015 0 : CHK_PTR_NULL(hcclComm);
1016 0 : if (!hcclComm->IsCommunicatorV2()) {
1017 0 : return HCCL_SUCCESS;
1018 : }
1019 0 : hccl::CollComm* collComm = hcclComm->GetCollComm();
1020 0 : CHK_PTR_NULL(collComm);
1021 0 : HcclCommDfx* hcclCommDfx = collComm->GetHcclCommDfx();
1022 0 : CHK_PTR_NULL(hcclCommDfx);
1023 :
1024 0 : auto* mirrorTaskMgr = hcclCommDfx->GetMirrorTaskManager();
1025 0 : CHK_PTR_NULL(mirrorTaskMgr);
1026 0 : if (mirrorTaskMgr->GetCurrDfxOpInfo() == nullptr) {
1027 0 : HCCL_INFO("[%s] commId[%s] currDfxOpInfo is null, skip report.", __func__, collComm->GetCommId().c_str());
1028 0 : return HCCL_SUCCESS;
1029 : }
1030 :
1031 0 : std::string kernelNameStr(kernelName);
1032 0 : uint32_t threadId = SalGetTid();
1033 0 : bool isBaseOpMode = false;
1034 0 : CHK_RET(hcclCommDfx->IsOpBase(isBaseOpMode));
1035 0 : CHK_RET(hcclCommDfx->ReportKernel(beginTime, collComm->GetCommId(), kernelNameStr, threadId, !isBaseOpMode));
1036 :
1037 0 : Hccl::TaskParam taskParam{};
1038 0 : taskParam.beginTime = beginTime;
1039 0 : taskParam.taskType = Hccl::TaskParamType::TASK_AICPU_KERNEL;
1040 0 : taskParam.endTime = Hccl::DlProfFunction::GetInstance().dlMsprofSysCycleTime();
1041 0 : uint32_t taskId = INVALID_UINT;
1042 0 : uint32_t streamId = INVALID_UINT;
1043 0 : CHK_RET(hrtGetTaskIdAndStreamID(taskId, streamId));
1044 0 : HCCL_INFO("[%s] taskId[%u], streamId[%u].", __func__, taskId, streamId);
1045 0 : hcclCommDfx->SetAicpuTaskIdAndStreamId(taskId, streamId);
1046 0 : CHK_RET(hcclCommDfx->AddTaskInfoCallback(streamId, taskId, taskParam, INVALID_U64));
1047 0 : HCCL_INFO("[HcclReportAicpuKernel] HcclReportAicpuKernel success");
1048 0 : return HCCL_SUCCESS;
1049 0 : }
1050 :
1051 1 : extern HcclResult HcclReportAivKernel(HcclComm comm, uint64_t beginTime)
1052 : {
1053 1 : HCCL_INFO("[%s] START, comm[%p].", __func__, comm);
1054 1 : CHK_PRT_RET(comm == nullptr, HCCL_ERROR("[%s] comm is null", __func__), HCCL_E_PTR);
1055 0 : auto hcclComm = static_cast<hccl::hcclComm*>(comm);
1056 0 : CHK_PTR_NULL(hcclComm);
1057 0 : if (!hcclComm->IsCommunicatorV2()) {
1058 0 : HCCL_ERROR("[%s] comm is not supported", __func__);
1059 0 : return HCCL_E_NOT_SUPPORT;
1060 : }
1061 0 : hccl::CollComm* collComm = hcclComm->GetCollComm();
1062 0 : CHK_PTR_NULL(collComm);
1063 0 : HcclCommDfx* hcclCommDfx = collComm->GetHcclCommDfx();
1064 0 : CHK_PTR_NULL(hcclCommDfx);
1065 :
1066 0 : Hccl::TaskParam taskParam{};
1067 0 : taskParam.beginTime = beginTime;
1068 0 : taskParam.taskType = Hccl::TaskParamType::TASK_AIV;
1069 0 : taskParam.endTime = Hccl::DlProfFunction::GetInstance().dlMsprofSysCycleTime();
1070 0 : taskParam.isMaster = true;
1071 0 : uint32_t taskId = INVALID_UINT;
1072 0 : uint32_t streamId = INVALID_UINT;
1073 0 : CHK_RET(hrtGetTaskIdAndStreamID(taskId, streamId));
1074 0 : CHK_RET(hcclCommDfx->AddTaskInfoCallback(streamId, taskId, taskParam, INVALID_U64));
1075 0 : HCCL_INFO("[HcclReportAivKernel] HcclReportAivKernel success");
1076 0 : return HCCL_SUCCESS;
1077 0 : }
1078 :
1079 0 : int32_t HcommChannelNotifyWaitOnThreadWithDefaultTimeout(ThreadHandle thread, ChannelHandle channel, uint32_t localNotifyIdx)
1080 : {
1081 0 : HCCL_ERROR("[%s] thread[0x%llx], channel[0x%llx], localNotifyIdx[%u] not support in cpu.",
1082 : __func__, thread, channel, localNotifyIdx);
1083 0 : return HCCL_E_NOT_SUPPORT;
1084 : }
1085 :
1086 0 : int32_t HcommThreadNotifyWaitOnThreadWithDefaultTimeout(ThreadHandle thread, uint32_t notifyIdx)
1087 : {
1088 0 : HCCL_INFO("[%s] data cpu START. thread[0x%llx], notifyIdx[%u].", __func__, thread, notifyIdx);
1089 :
1090 : uint32_t notifyWaitTimeout;
1091 0 : g_threadLaunchCtx.GetNotifyWaitTimeOut(notifyWaitTimeout);
1092 :
1093 0 : HCCL_DEBUG("[%s] data cpu Using default timeout: %u s", __func__, notifyWaitTimeout);
1094 :
1095 0 : int32_t ret = HcommThreadNotifyWaitOnThread(thread, notifyIdx, notifyWaitTimeout);
1096 0 : if (ret != HCCL_SUCCESS) {
1097 0 : HCCL_ERROR("[%s] data cpu HcommThreadNotifyWaitOnThread FAILED. thread[0x%llx], notifyIdx[%u], ret[%d]",
1098 : __func__, thread, notifyIdx, ret);
1099 0 : return ret;
1100 : }
1101 :
1102 0 : HCCL_INFO("[%s] data cpu SUCCESS.", __func__);
1103 0 : return HCCL_SUCCESS;
1104 : }
|
