Line data Source code
1 : /**
2 : * Copyright (c) 2026 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 <thread>
12 : #include <sstream>
13 : #include <string>
14 : #include <vector>
15 :
16 : #include "hccl/hccl_launch.h"
17 : #include "acl/acl_rt.h"
18 : #include "hccl_group.h"
19 : #include "hccl_res_expt.h"
20 : #include "hccl_aicpu_interface.h"
21 :
22 : #include "hccl_independent_common.h"
23 : #include "group_schedule_mgr.h"
24 :
25 : using namespace hccl;
26 :
27 : constexpr uint32_t NUM_ZERO = 0;
28 : constexpr uint32_t NUM_ONE = 1;
29 : constexpr uint32_t NUM_TWO = 2;
30 : constexpr uint32_t NUM_THREE = 3;
31 : static uint32_t g_KernelLaunchTimeout = UINT16_MAX;
32 :
33 : static HcclResult
34 3 : LaunchNotifyWaitToThread(HcclComm comm, aclrtStream unfoldStream, ThreadHandle srcThread, uint32_t dstNotifyIdx)
35 : {
36 : aclrtFuncHandle funcHandle;
37 : aclrtArgsHandle argsHandle;
38 3 : std::string kernelName = "RunAicpuNotifyWait";
39 : // 1. 获取 function handle
40 3 : hccl::hcclComm* hcclComm = static_cast<hccl::hcclComm*>(comm);
41 3 : auto binKernelHandle = hcclComm->GetBinHandle();
42 3 : aclError ret = aclrtBinaryGetFunction(binKernelHandle, kernelName.c_str(), &funcHandle);
43 3 : CHK_PRT_RET(
44 : ret != ACL_SUCCESS,
45 : HCCL_ERROR(
46 : "[aclrtBinaryGetFunction]errNo[0x%016llx] get func handle failed, "
47 : "kernelName:%s",
48 : ret, kernelName.c_str()),
49 : HCCL_E_RUNTIME);
50 :
51 : // 2. 初始化 args handle
52 3 : ret = aclrtKernelArgsInit(funcHandle, &argsHandle);
53 3 : CHK_PRT_RET(
54 : ret != ACL_SUCCESS,
55 : HCCL_ERROR("[aclrtKernelArgsInit]errNo[0x%016llx] args init failed, kernelName:%s", ret, kernelName.c_str()),
56 : HCCL_E_RUNTIME);
57 :
58 : // 3. 准备参数并 append
59 : ThreadNotifyWaitParam param;
60 3 : param.thread = srcThread;
61 3 : param.notifyIdx = dstNotifyIdx;
62 : aclrtParamHandle paraHandle;
63 3 : ret = aclrtKernelArgsAppend(argsHandle, ¶m, sizeof(ThreadNotifyWaitParam), ¶Handle);
64 3 : CHK_PRT_RET(
65 : ret != ACL_SUCCESS,
66 : HCCL_ERROR(
67 : "[aclrtKernelArgsAppend]errNo[0x%016llx] args append failed, kernelName:%s", ret, kernelName.c_str()),
68 : HCCL_E_RUNTIME);
69 :
70 : // 4. finalize args
71 3 : ret = aclrtKernelArgsFinalize(argsHandle);
72 3 : CHK_PRT_RET(
73 : ret != ACL_SUCCESS,
74 : HCCL_ERROR(
75 : "[aclrtKernelArgsFinalize]errNo[0x%016llx] args finalize failed, kernelName:%s", ret, kernelName.c_str()),
76 : HCCL_E_RUNTIME);
77 :
78 : // 5. 下发 kernel
79 : aclrtLaunchKernelCfg cfg;
80 : aclrtLaunchKernelAttr attr;
81 3 : attr.id = ACL_RT_LAUNCH_KERNEL_ATTR_TIMEOUT;
82 3 : attr.value.timeout = g_KernelLaunchTimeout;
83 3 : cfg.numAttrs = 1;
84 3 : cfg.attrs = &attr;
85 3 : constexpr u32 numBlocks = 1;
86 :
87 3 : ret = aclrtLaunchKernelWithConfig(funcHandle, numBlocks, unfoldStream, &cfg, argsHandle, nullptr);
88 3 : CHK_PRT_RET(
89 : ret != ACL_SUCCESS,
90 : HCCL_ERROR(
91 : "[aclrtLaunchKernelWithConfig]errNo[0x%016llx] launch kernel failed, "
92 : "kernelName:%s",
93 : ret, kernelName.c_str()),
94 : HCCL_E_RUNTIME);
95 :
96 3 : return HCCL_SUCCESS;
97 3 : }
98 :
99 3 : static HcclResult LaunchP2pExec(
100 : HcclComm comm, aclrtStream unfoldStream, const HcclKernelFuncInfo* funcInfo, const void* funcArgs, uint32_t argSize,
101 : ThreadHandle sendRecvThread)
102 : {
103 : aclrtFuncHandle funcHandle;
104 : aclrtArgsHandle argsHandle;
105 : // 1. 获取 function handle
106 3 : aclrtBinHandle binKernelHandle = nullptr;
107 3 : hccl::hcclComm* hcclComm = static_cast<hccl::hcclComm*>(comm);
108 3 : CollComm* collComm = hcclComm->GetCollComm();
109 3 : CHK_PTR_NULL(collComm);
110 2 : CHK_RET(collComm->GetHcclBinHandle(binKernelHandle));
111 2 : aclError ret = aclrtBinaryGetFunction(binKernelHandle, funcInfo->kernelFuncName, &funcHandle);
112 2 : CHK_PRT_RET(
113 : ret != ACL_SUCCESS,
114 : HCCL_ERROR(
115 : "[aclrtBinaryGetFunction]errNo[0x%016llx] get func handle failed, kernelName:%s", ret,
116 : funcInfo->kernelFuncName),
117 : HCCL_E_RUNTIME);
118 :
119 : // 2. 初始化 args handle
120 2 : ret = aclrtKernelArgsInit(funcHandle, &argsHandle);
121 2 : CHK_PRT_RET(
122 : ret != ACL_SUCCESS,
123 : HCCL_ERROR(
124 : "[aclrtKernelArgsInit]errNo[0x%016llx] args init failed, kernelName:%s", ret, funcInfo->kernelFuncName),
125 : HCCL_E_RUNTIME);
126 :
127 : // 3. 准备参数并 append
128 : HcclP2pKernelParam params;
129 2 : params.sendRecvThread = sendRecvThread;
130 2 : memset_s(params.opParams, P2P_MAX_ARG_SIZE, 0, P2P_MAX_ARG_SIZE);
131 2 : memcpy_s(params.opParams, P2P_MAX_ARG_SIZE, funcArgs, argSize);
132 :
133 : aclrtParamHandle paraHandle;
134 2 : ret = aclrtKernelArgsAppend(argsHandle, ¶ms, sizeof(HcclP2pKernelParam), ¶Handle);
135 2 : CHK_PRT_RET(
136 : ret != ACL_SUCCESS,
137 : HCCL_ERROR(
138 : "[aclrtKernelArgsAppend]errNo[0x%016llx] args append failed, kernelName:%s", ret, funcInfo->kernelFuncName),
139 : HCCL_E_RUNTIME);
140 :
141 : // 4. finalize args
142 2 : ret = aclrtKernelArgsFinalize(argsHandle);
143 2 : CHK_PRT_RET(
144 : ret != ACL_SUCCESS,
145 : HCCL_ERROR(
146 : "[aclrtKernelArgsFinalize]errNo[0x%016llx] args finalize failed, kernelName:%s", ret,
147 : funcInfo->kernelFuncName),
148 : HCCL_E_RUNTIME);
149 :
150 : // 5. 下发 kernel
151 : aclrtLaunchKernelCfg cfg;
152 : aclrtLaunchKernelAttr attr;
153 2 : attr.id = ACL_RT_LAUNCH_KERNEL_ATTR_TIMEOUT;
154 2 : attr.value.timeout = g_KernelLaunchTimeout;
155 2 : cfg.numAttrs = 1;
156 2 : cfg.attrs = &attr;
157 2 : constexpr u32 numBlocks = 1;
158 :
159 2 : ret = aclrtLaunchKernelWithConfig(funcHandle, numBlocks, unfoldStream, &cfg, argsHandle, nullptr);
160 2 : CHK_PRT_RET(
161 : ret != ACL_SUCCESS,
162 : HCCL_ERROR(
163 : "[aclrtLaunchKernelWithConfig]errNo[0x%016llx] launch kernel failed, "
164 : "kernelName:%s",
165 : ret, funcInfo->kernelFuncName),
166 : HCCL_E_RUNTIME);
167 :
168 2 : return HCCL_SUCCESS;
169 : }
170 :
171 : // 放到kernel launch的地方
172 2 : static HcclResult LaunchNotifyRecordToThread(
173 : HcclComm comm, aclrtStream unfoldStream, ThreadHandle srcThread, ThreadHandle dstThread, uint32_t dstNotifyIdx)
174 : {
175 : aclrtFuncHandle funcHandle;
176 : aclrtArgsHandle argsHandle;
177 2 : std::string kernelName = "RunAicpuNotifyRecord";
178 :
179 : // 1. 获取 function handle
180 2 : hccl::hcclComm* hcclComm = static_cast<hccl::hcclComm*>(comm);
181 2 : auto binKernelHandle = hcclComm->GetBinHandle();
182 2 : aclError ret = aclrtBinaryGetFunction(binKernelHandle, kernelName.c_str(), &funcHandle);
183 2 : CHK_PRT_RET(
184 : ret != ACL_SUCCESS,
185 : HCCL_ERROR(
186 : "[aclrtBinaryGetFunction]errNo[0x%016llx] get func handle failed, kernelName:%s", ret, kernelName.c_str()),
187 : HCCL_E_RUNTIME);
188 :
189 : // 2. 初始化 args handle
190 2 : ret = aclrtKernelArgsInit(funcHandle, &argsHandle);
191 2 : CHK_PRT_RET(
192 : ret != ACL_SUCCESS,
193 : HCCL_ERROR("[aclrtKernelArgsInit]errNo[0x%016llx] args init failed, kernelName:%s", ret, kernelName.c_str()),
194 : HCCL_E_RUNTIME);
195 :
196 : // 3. 准备参数并 append
197 : ThreadNotifyRecordParam param;
198 2 : param.thread = srcThread;
199 2 : param.dstThread = dstThread;
200 2 : param.dstNotifyIdx = dstNotifyIdx;
201 : aclrtParamHandle paraHandle;
202 2 : ret = aclrtKernelArgsAppend(argsHandle, ¶m, sizeof(ThreadNotifyRecordParam), ¶Handle);
203 2 : CHK_PRT_RET(
204 : ret != ACL_SUCCESS,
205 : HCCL_ERROR(
206 : "[aclrtKernelArgsAppend]errNo[0x%016llx] args append failed, kernelName:%s", ret, kernelName.c_str()),
207 : HCCL_E_RUNTIME);
208 :
209 : // 4. finalize args
210 2 : ret = aclrtKernelArgsFinalize(argsHandle);
211 2 : CHK_PRT_RET(
212 : ret != ACL_SUCCESS,
213 : HCCL_ERROR(
214 : "[aclrtKernelArgsFinalize]errNo[0x%016llx] args finalize failed, kernelName:%s", ret, kernelName.c_str()),
215 : HCCL_E_RUNTIME);
216 :
217 : // 5. 下发 kernel
218 : aclrtLaunchKernelCfg cfg;
219 : aclrtLaunchKernelAttr attr;
220 2 : attr.id = ACL_RT_LAUNCH_KERNEL_ATTR_TIMEOUT;
221 2 : attr.value.timeout = g_KernelLaunchTimeout;
222 2 : cfg.numAttrs = 1;
223 2 : cfg.attrs = &attr;
224 2 : constexpr u32 numBlocks = 1;
225 :
226 2 : ret = aclrtLaunchKernelWithConfig(funcHandle, numBlocks, unfoldStream, &cfg, argsHandle, nullptr);
227 2 : CHK_PRT_RET(
228 : ret != ACL_SUCCESS,
229 : HCCL_ERROR(
230 : "[aclrtLaunchKernelWithConfig]errNo[0x%016llx] launch kernel failed, "
231 : "kernelName:%s",
232 : ret, kernelName.c_str()),
233 : HCCL_E_RUNTIME);
234 :
235 2 : return HCCL_SUCCESS;
236 2 : }
237 :
238 1 : static HcclResult AicpuKernelLaunchDirect(
239 : HcclComm comm, const HcclKernelFuncInfo* funcInfo, ThreadHandle aicpuThreadHandle, aclrtStream unfoldStream,
240 : aclrtStream userStream)
241 : {
242 1 : CHK_PTR_NULL(comm);
243 1 : CHK_PTR_NULL(unfoldStream);
244 1 : CHK_PTR_NULL(userStream);
245 1 : CHK_PTR_NULL(funcInfo);
246 :
247 1 : void* args = funcInfo->args;
248 1 : uint32_t argSize = funcInfo->argSize;
249 1 : if (argSize > 0 && args == nullptr) {
250 0 : HCCL_ERROR("[AicpuKernelLaunchDirect] args is null but argSize[%u] > 0", argSize);
251 0 : return HCCL_E_PTR;
252 : }
253 :
254 1 : ThreadHandle cpuTsThread{0};
255 1 : ThreadHandle exportedAicpuTsThread{0};
256 1 : ThreadHandle exportedCpuTsThread{0};
257 : uint32_t notifyNumOnMainThread;
258 1 : CHK_RET(HcclThreadAcquireWithStream(comm, COMM_ENGINE_CPU_TS, userStream, NUM_THREE, &cpuTsThread));
259 1 : CHK_RET(HcclThreadExportToCommEngine(comm, NUM_ONE, &cpuTsThread, COMM_ENGINE_AICPU_TS, &exportedAicpuTsThread));
260 1 : CHK_RET(HcclThreadExportToCommEngine(comm, NUM_ONE, &aicpuThreadHandle, COMM_ENGINE_CPU_TS, &exportedCpuTsThread));
261 1 : CHK_RET(HcclGetNotifyNumInThread(comm, exportedCpuTsThread, COMM_ENGINE_AICPU_TS, ¬ifyNumOnMainThread));
262 :
263 1 : CHK_RET(static_cast<HcclResult>(
264 : HcommThreadNotifyRecordOnThread(cpuTsThread, exportedCpuTsThread, notifyNumOnMainThread - 1))); // h2d record
265 1 : uint64_t beginTime = HcommGetProfilingSysCycleTime(); // AicpuKernel report start
266 1 : CHK_RET(LaunchNotifyWaitToThread(comm, unfoldStream, aicpuThreadHandle, notifyNumOnMainThread - 1)); // device wait
267 1 : CHK_RET(LaunchP2pExec(comm, unfoldStream, funcInfo, args, argSize, aicpuThreadHandle)); // device run task
268 0 : CHK_RET(LaunchNotifyRecordToThread(
269 : comm, unfoldStream, aicpuThreadHandle, exportedAicpuTsThread, NUM_ZERO)); // d2h record
270 0 : std::string kernelNameCStr(funcInfo->kernelFuncName);
271 0 : HcclResult ret = HcclReportAicpuKernel(comm, beginTime, kernelNameCStr.data()); // AicpuKernel report end
272 0 : if (ret != HCCL_SUCCESS) {
273 0 : HCCL_ERROR(
274 : "[AicpuKernelLaunchDirect] HcclReportAicpuKernel failed, beginTime %llu, kernelName %s, ret %d ", beginTime,
275 : kernelNameCStr.c_str(), ret);
276 0 : return ret;
277 : }
278 0 : CHK_RET(
279 : static_cast<HcclResult>(HcommThreadNotifyWaitOnThreadWithDefaultTimeout(cpuTsThread, NUM_ZERO))); // host wait
280 :
281 0 : return HCCL_SUCCESS;
282 0 : }
283 :
284 7 : HcclResult HcclAicpuKernelLaunch(
285 : HcclComm comm, const HcclOpDesc* opInfo, const HcclKernelFuncInfo* funcInfo, ThreadHandle aicpuThreadHandle,
286 : aclrtStream userStream, const HcclKernelLaunchCfg* kernelLaunchCfg)
287 : {
288 7 : CHK_PTR_NULL(comm);
289 6 : CHK_PTR_NULL(userStream);
290 5 : CHK_PTR_NULL(funcInfo);
291 4 : CHK_PTR_NULL(opInfo);
292 3 : CHK_PTR_NULL(kernelLaunchCfg);
293 :
294 2 : uint32_t argSize = funcInfo->argSize;
295 2 : void* args = funcInfo->args;
296 :
297 2 : g_KernelLaunchTimeout = kernelLaunchCfg->timeOut;
298 2 : if (argSize > 0 && args == nullptr) {
299 1 : HCCL_ERROR("[HcclAicpuKernelLaunch] args is null but argSize[%u] > 0", argSize);
300 1 : return HCCL_E_PTR;
301 : }
302 :
303 1 : HCCL_INFO(
304 : "[HcclAicpuKernelLaunch] opDescType[%u], kernelSo[%s], kernelFuncName[%s], argSize[%u], "
305 : "aicpuThreadHandle[%llu], hcclGroupDepth[%d]",
306 : opInfo->opDescType, funcInfo->kernelSoName, funcInfo->kernelFuncName, argSize, aicpuThreadHandle,
307 : hcclGroupDepth);
308 :
309 1 : if (hcclGroupDepth > 0) {
310 0 : hccl::hcclComm* hcclComm = static_cast<hccl::hcclComm*>(comm);
311 0 : CollComm* collComm = hcclComm->GetCollComm();
312 0 : CHK_PTR_NULL(collComm);
313 0 : if (argSize > P2P_MAX_ARG_SIZE) {
314 0 : HCCL_ERROR("[HcclAicpuKernelLaunch] argSize[%u] over P2P_MAX_ARG_SIZE", argSize);
315 0 : return HCCL_E_PARA;
316 : }
317 0 : HCCL_INFO("[HcclAicpuKernelLaunch] group mode, add p2p task hcclGroupDepth[%d]", hcclGroupDepth);
318 : HcclP2pTask task;
319 0 : task.desc = opInfo->p2p;
320 0 : task.stream = opInfo->p2p.unfoldStream;
321 0 : memcpy_s(
322 0 : task.funcInfo.kernelSoName, HCCL_KERNEL_SO_NAME_MAX_LEN, funcInfo->kernelSoName,
323 : HCCL_KERNEL_SO_NAME_MAX_LEN);
324 0 : memcpy_s(
325 0 : task.funcInfo.kernelFuncName, HCCL_KERNEL_FUNC_NAME_MAX_LEN, funcInfo->kernelFuncName,
326 : HCCL_KERNEL_FUNC_NAME_MAX_LEN);
327 0 : memcpy_s(task.args, P2P_MAX_ARG_SIZE, args, argSize);
328 0 : task.argSize = argSize;
329 0 : task.usrStream = userStream;
330 0 : CHK_RET(collComm->groupScheduleMgr->AppendGroupP2pTask(comm, task, opInfo->p2p));
331 0 : return HCCL_SUCCESS;
332 : }
333 :
334 1 : return AicpuKernelLaunchDirect(comm, funcInfo, aicpuThreadHandle, opInfo->p2p.unfoldStream, userStream);
335 : }
336 :
337 1 : static HcclResult GetStreams(
338 : const CollComm* collComm, const std::vector<HcclP2pTask>& sortedSendQue,
339 : const std::vector<HcclP2pTask>& sortedRecvQue, aclrtStream& unfoldStream, const aclrtStream& usrStream)
340 : {
341 1 : if (!sortedSendQue.empty()) {
342 1 : unfoldStream = sortedSendQue[0].stream;
343 1 : CHK_RET(collComm->groupScheduleMgr->SetUsrStream(sortedSendQue[0].usrStream));
344 0 : } else if (!sortedRecvQue.empty()) {
345 0 : unfoldStream = sortedRecvQue[0].stream;
346 0 : CHK_RET(collComm->groupScheduleMgr->SetUsrStream(sortedRecvQue[0].usrStream));
347 : } else {
348 0 : return HCCL_E_INTERNAL;
349 : }
350 1 : return HCCL_SUCCESS;
351 : }
352 :
353 2 : HcclResult groupLaunchA5()
354 : {
355 2 : std::vector<HcclComm> hcclGroupCommListV2 = GetHcclGroupCommList();
356 2 : HCCL_INFO("[groupLaunchA5] to the start hcclGroupCommListV2.size[%zu]", hcclGroupCommListV2.size());
357 :
358 3 : for (HcclComm comm : hcclGroupCommListV2) {
359 1 : hccl::hcclComm* hcclComm = static_cast<hccl::hcclComm*>(comm);
360 1 : CollComm* collComm = hcclComm->GetCollComm();
361 1 : CHK_PTR_NULL(collComm);
362 :
363 : /*新建send/recv流*/
364 : ThreadHandle sendRecv[2];
365 1 : CHK_RET(HcclThreadAcquire(comm, COMM_ENGINE_AICPU_TS, NUM_TWO, NUM_ONE, sendRecv));
366 1 : ThreadHandle aicpuSendThread = sendRecv[0], aicpuRecvThread = sendRecv[1];
367 1 : std::vector<HcclP2pTask> sortedSendQue, sortedRecvQue;
368 1 : CHK_RET(collComm->groupScheduleMgr->GetP2pTaskSchedule(sortedSendQue, sortedRecvQue));
369 1 : aclrtStream unfoldStream = nullptr, usrStream = nullptr;
370 1 : ThreadHandle cpuTsThread = 0, exportedAicpuTsThread = 0, exportedCpuTsSendThread = 0,
371 1 : exportedCpuTsRecvThread = 0;
372 1 : CHK_RET(GetStreams(collComm, sortedSendQue, sortedRecvQue, unfoldStream, usrStream));
373 :
374 1 : CHK_RET(collComm->groupScheduleMgr->GetUsrStream(usrStream));
375 1 : CHK_RET(HcclThreadAcquireWithStream(comm, COMM_ENGINE_CPU_TS, usrStream, NUM_THREE, &cpuTsThread));
376 1 : CHK_RET(
377 : HcclThreadExportToCommEngine(comm, NUM_ONE, &cpuTsThread, COMM_ENGINE_AICPU_TS, &exportedAicpuTsThread));
378 1 : CHK_RET(HcclThreadExportToCommEngine(
379 : comm, NUM_ONE, &aicpuSendThread, COMM_ENGINE_CPU_TS, &exportedCpuTsSendThread));
380 1 : CHK_RET(HcclThreadExportToCommEngine(
381 : comm, NUM_ONE, &aicpuRecvThread, COMM_ENGINE_CPU_TS, &exportedCpuTsRecvThread));
382 1 : CHK_RET(
383 : static_cast<HcclResult>(HcommThreadNotifyRecordOnThread(cpuTsThread, exportedCpuTsSendThread, NUM_ZERO)));
384 1 : CHK_RET(
385 : static_cast<HcclResult>(HcommThreadNotifyRecordOnThread(cpuTsThread, exportedCpuTsRecvThread, NUM_ZERO)));
386 :
387 : // 下发wait kernel
388 1 : CHK_RET(LaunchNotifyWaitToThread(comm, unfoldStream, aicpuSendThread, NUM_ZERO));
389 1 : CHK_RET(LaunchNotifyWaitToThread(comm, unfoldStream, aicpuRecvThread, NUM_ZERO));
390 :
391 : // Send/Recv交替执行以避免死锁
392 2 : for (size_t sendIdx = 0, recvIdx = 0; sendIdx < sortedSendQue.size() || recvIdx < sortedRecvQue.size();) {
393 1 : if (sendIdx < sortedSendQue.size()) {
394 1 : CHK_RET(LaunchP2pExec(
395 : comm, sortedSendQue[sendIdx].stream, &sortedSendQue[sendIdx].funcInfo, sortedSendQue[sendIdx].args,
396 : sortedSendQue[sendIdx].argSize, aicpuSendThread));
397 1 : sendIdx++;
398 : }
399 1 : if (recvIdx < sortedRecvQue.size()) {
400 1 : CHK_RET(LaunchP2pExec(
401 : comm, sortedRecvQue[recvIdx].stream, &sortedRecvQue[recvIdx].funcInfo, sortedRecvQue[recvIdx].args,
402 : sortedRecvQue[recvIdx].argSize, aicpuRecvThread));
403 1 : recvIdx++;
404 : }
405 : }
406 :
407 : // 下发record kernel
408 1 : CHK_RET(LaunchNotifyRecordToThread(comm, unfoldStream, aicpuSendThread, exportedAicpuTsThread, NUM_ONE));
409 1 : CHK_RET(LaunchNotifyRecordToThread(comm, unfoldStream, aicpuRecvThread, exportedAicpuTsThread, NUM_TWO));
410 :
411 1 : CHK_RET(static_cast<HcclResult>(HcommThreadNotifyWaitOnThreadWithDefaultTimeout(cpuTsThread, NUM_ONE)));
412 1 : CHK_RET(static_cast<HcclResult>(HcommThreadNotifyWaitOnThreadWithDefaultTimeout(cpuTsThread, NUM_TWO)));
413 1 : }
414 :
415 2 : SetHcclP2pTaskNums(0);
416 2 : ClearHcclGroupCommList();
417 :
418 2 : return HCCL_SUCCESS;
419 2 : }
|
