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