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