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 "task_orchestrator.h"
12 : #include <cmath>
13 : #include "common/aicpu_sqe_context.h"
14 : #include "common/aicpu_hccl_common.h"
15 : #include "dfx/mc2_trace_utils.h"
16 : #include "utils/hccl_aicpu_utils.h"
17 : #include "common/aicpu_kfc_utils.h"
18 : #include "framework/aicpu_kfc_prof.h"
19 : #include "log.h"
20 : #include "utils/aicpu_hdc_utils.h"
21 : #include "aicpu_operator_pub.h"
22 : #include "hccl_types.h"
23 : #include "aicpu_allgather.h"
24 : #include "aicpu_reduce_scatter.h"
25 : #include "aicpu_dmy_cal_allreduce.h"
26 : #include "aicpu_allreduce.h"
27 : #include "aicpu_alltoall.h"
28 :
29 : using namespace hccl;
30 : namespace {
31 : #define KFC_GET_START_TIME() \
32 : ((AicpuKfcUtils::NeedRecordTimeTaken(*AicpuGetComContext())) ? GetCurCpuTimestamp() : 0)
33 :
34 : #define RECORD_FILL_SQE_TIME(START_TIME) \
35 : do { \
36 : AicpuComContext *commctx__ = AicpuGetComContext(); \
37 : if (!AicpuKfcUtils::NeedRecordTimeTaken(*commctx__)) { break; } \
38 : AicpuKfcProf::GetProInst(*commctx__).fillSqeTimes += GetCurCpuTimestamp() - (START_TIME); \
39 : } while (0)
40 :
41 : #define RECORD_PROF_TIME(VAR) \
42 : do { \
43 : AicpuComContext *commctx__ = AicpuGetComContext(); \
44 : if (!AicpuKfcUtils::NeedRecordTimeTaken(*commctx__)) { break; } \
45 : uint32_t recordIndex = AicpuKfcProf::GetProInst(*commctx__).workCnt; \
46 : recordIndex = (recordIndex >= AC_MAX_PROF_COMM_CNT) ? (AC_MAX_PROF_COMM_CNT - 1) : recordIndex; \
47 : AicpuKfcProf::GetProInst(*commctx__).commLoop[recordIndex].VAR = GetCurCpuTimestamp(true); \
48 : } while (0)
49 : }
50 :
51 55 : HcclResult TaskOrchestrator::DoPreSync()
52 : {
53 : // 15 sqe on main, 35 sqe on sub
54 55 : CHK_RET(MainSubPreSync());
55 :
56 55 : CHK_RET(IpcPreSync());
57 :
58 55 : CHK_RET(MainSubPostSync());
59 :
60 55 : CHK_RET(MainSubPreSync());
61 :
62 55 : HCCL_INFO("[SQE]Do pre sync on main stream 21 tasks, sub stream 35 tasks");
63 55 : return HCCL_SUCCESS;
64 : }
65 :
66 55 : HcclResult TaskOrchestrator::DoPostSync()
67 : {
68 : // 8 sqe on main, 21 sqe on sub
69 55 : CHK_RET(IpcPostSync());
70 :
71 55 : CHK_RET(MainSubPostSync());
72 :
73 55 : HCCL_INFO("[SQE]Do post sync on main stream 7 tasks, sub stream 21 tasks");
74 55 : return HCCL_SUCCESS;
75 : }
76 :
77 105 : HcclResult TaskOrchestrator::SelfCpySnd2Win(void *sndAddr, u64 dataSize, u64 sndOffset, u64 winOffset,
78 : HcclReduceOp opType, HcclDataType dataType)
79 : {
80 105 : const u64 startTime = KFC_GET_START_TIME();
81 105 : auto ctx = AicpuGetComContext();
82 105 : u32 rankId = ctx->rankId;
83 :
84 105 : AicpuComRankInfo *rankInfo = &ctx->rankInfo[rankId];
85 105 : void *src = static_cast<void *>(static_cast<s8 *>(sndAddr) + sndOffset);
86 105 : void *dst = reinterpret_cast<void *>(static_cast<const uintptr_t>(rankInfo->window) + winOffset);
87 :
88 105 : CHK_RET(AicpuDispatcher::CopyData(rankId, src, dst, dataSize, dataType, opType, rankId));
89 :
90 104 : RECORD_FILL_SQE_TIME(startTime);
91 104 : return HCCL_SUCCESS;
92 : }
93 :
94 1 : HcclResult TaskOrchestrator::SelfCpyRcv2Win(void *rcvAddr, u64 dataSize, u64 rcvOffset, u64 winOffset,
95 : HcclReduceOp opType, HcclDataType dataType)
96 : {
97 1 : const u64 startTime = KFC_GET_START_TIME();
98 1 : auto ctx = AicpuGetComContext();
99 1 : u32 rankId = ctx->rankId;
100 :
101 1 : AicpuComRankInfo *rankInfo = &ctx->rankInfo[rankId];
102 1 : void *src = static_cast<void *>(static_cast<s8 *>(rcvAddr) + rcvOffset);
103 1 : void *dst = reinterpret_cast<void *>(static_cast<const uintptr_t>(rankInfo->window) + winOffset);
104 :
105 1 : CHK_RET(AicpuDispatcher::CopyData(rankId, src, dst, dataSize, dataType, opType, rankId));
106 :
107 1 : RECORD_FILL_SQE_TIME(startTime);
108 1 : return HCCL_SUCCESS;
109 : }
110 :
111 2 : HcclResult TaskOrchestrator::IpcCpyWin2Win(u64 *dataSize, u64 *winOffsets, HcclReduceOp opType, u64 sendOff,
112 : HcclDataType dataType)
113 : {
114 2 : const u64 startTime = KFC_GET_START_TIME();
115 2 : auto ctx = AicpuGetComContext();
116 2 : AicpuComRankInfo *selfRankInfo = &ctx->rankInfo[ctx->rankId];
117 2 : u64 offset = (winOffsets == nullptr) ? 0 : winOffsets[ctx->rankId];
118 2 : void *selfWindow = reinterpret_cast<void *>(static_cast<const uintptr_t>(selfRankInfo->window));
119 2 : void *dst = static_cast<void *>(static_cast<s8 *>(selfWindow) + sendOff + offset);
120 5 : for (u32 index = 0; index < ctx->rankNum; index++) {
121 3 : if (index != ctx->rankId) {
122 1 : AicpuComRankInfo *rankInfo = &ctx->rankInfo[index];
123 1 : void *otherRankWindow = reinterpret_cast<void *>(static_cast<const uintptr_t>(rankInfo->window));
124 1 : void *src = static_cast<void *>(static_cast<s8 *>(otherRankWindow) + sendOff + offset);
125 1 : CHK_RET(AicpuDispatcher::CopyData(index, src, dst, dataSize[ctx->rankId], dataType, opType, index));
126 : }
127 : }
128 2 : RECORD_FILL_SQE_TIME(startTime);
129 2 : return HCCL_SUCCESS;
130 : }
131 :
132 1 : HcclResult TaskOrchestrator::IpcCpyWin2Win(const std::vector<u64> &dataSizes, u64 sendOff,
133 : const std::vector<u64> &winOffsets, HcclReduceOp opType, HcclDataType dataType)
134 : {
135 1 : const u64 startTime = KFC_GET_START_TIME();
136 1 : auto ctx = AicpuGetComContext();
137 1 : AicpuComRankInfo *selfRankInfo = &ctx->rankInfo[ctx->rankId];
138 1 : u64 offset = winOffsets.empty() ? 0 : winOffsets[ctx->rankId];
139 1 : void *selfWindow = reinterpret_cast<void *>(static_cast<const uintptr_t>(selfRankInfo->window));
140 1 : void *dst = static_cast<void *>(static_cast<s8 *>(selfWindow) + sendOff + offset);
141 3 : for (u32 index = 0; index < ctx->rankNum; index++) {
142 2 : if (index != ctx->rankId) {
143 1 : AicpuComRankInfo *rankInfo = &ctx->rankInfo[index];
144 1 : void *otherRankWindow = reinterpret_cast<void *>(static_cast<const uintptr_t>(rankInfo->window));
145 1 : void *src = static_cast<void *>(static_cast<s8 *>(otherRankWindow) + sendOff + offset);
146 1 : u64 dataSize = dataSizes.empty() ? 0 : dataSizes[ctx->rankId];
147 1 : CHK_RET(AicpuDispatcher::CopyData(index, src, dst, dataSize, dataType, opType, index));
148 : }
149 : }
150 1 : RECORD_FILL_SQE_TIME(startTime);
151 :
152 1 : return HCCL_SUCCESS;
153 : }
154 :
155 2 : HcclResult TaskOrchestrator::IpcCpyWin2WinEx(u32 mainRankId, u64 dataSize, u64 winOffset, HcclReduceOp opType,
156 : HcclDataType dataType, u32 maxStreamNum)
157 : {
158 2 : if (maxStreamNum == 0) {
159 1 : HCCL_ERROR("max stream num can not be zero");
160 1 : return HCCL_E_PARA;
161 : }
162 1 : const u64 startTime = KFC_GET_START_TIME();
163 1 : auto ctx = AicpuGetComContext();
164 1 : u32 rankId = ctx->rankId;
165 :
166 1 : AicpuComRankInfo *mainRankInfo = &ctx->rankInfo[mainRankId];
167 1 : AicpuComRankInfo *rankInfo = &ctx->rankInfo[rankId];
168 1 : void *src = reinterpret_cast<void *>(static_cast<const uintptr_t>(rankInfo->window) + winOffset);
169 1 : void *dst = reinterpret_cast<void *>(static_cast<const uintptr_t>(mainRankInfo->window) + winOffset);
170 :
171 1 : CHK_RET(AicpuDispatcher::CopyData(rankId % maxStreamNum, src, dst, dataSize, dataType, opType, mainRankId));
172 :
173 1 : RECORD_FILL_SQE_TIME(startTime);
174 1 : return HCCL_SUCCESS;
175 : }
176 :
177 36 : HcclResult TaskOrchestrator::SelfCpySnd2WinEx(u32 mainRankId, void *sndAddr, u64 dataSize, u64 sndOffset, u64 winOffset,
178 : HcclReduceOp opType, HcclDataType dataType, u32 maxStreamNum)
179 : {
180 36 : if (maxStreamNum == 0) {
181 2 : HCCL_ERROR("max stream num can not be zero");
182 2 : return HCCL_E_PARA;
183 : }
184 34 : const u64 startTime = KFC_GET_START_TIME();
185 34 : auto ctx = AicpuGetComContext();
186 34 : u32 rankId = ctx->rankId;
187 :
188 34 : AicpuComRankInfo *rankInfo = &ctx->rankInfo[mainRankId];
189 34 : void *src = static_cast<void *>(static_cast<s8 *>(sndAddr) + sndOffset);
190 34 : void *dst = reinterpret_cast<void *>(static_cast<const uintptr_t>(rankInfo->window) + winOffset);
191 :
192 34 : CHK_RET(AicpuDispatcher::CopyData(rankId % maxStreamNum, src, dst, dataSize, dataType, opType, mainRankId));
193 :
194 34 : RECORD_FILL_SQE_TIME(startTime);
195 34 : return HCCL_SUCCESS;
196 : }
197 :
198 25 : HcclResult TaskOrchestrator::SelfCpyWin2Rcv(void *rcvAddr, u64 dataSize, u64 winOffset, u64 rcvOffset,
199 : HcclReduceOp opType, HcclDataType dataType)
200 : {
201 25 : const u64 startTime = KFC_GET_START_TIME();
202 25 : auto ctx = AicpuGetComContext();
203 25 : u32 rankId = ctx->rankId;
204 25 : AicpuComRankInfo *rankInfo = &ctx->rankInfo[rankId];
205 :
206 25 : void *src = reinterpret_cast<void *>(static_cast<const uintptr_t>(rankInfo->window) + winOffset);
207 25 : void *dst = static_cast<void *>(static_cast<s8 *>(rcvAddr) + rcvOffset);
208 :
209 25 : CHK_RET(AicpuDispatcher::CopyData(rankId, src, dst, dataSize, dataType, opType, rankId));
210 :
211 24 : RECORD_FILL_SQE_TIME(startTime);
212 24 : return HCCL_SUCCESS;
213 : }
214 :
215 2 : HcclResult TaskOrchestrator::SelfCpyWin2RcvEx1(void *rcvAddr, u64 dataSize, u64 rcvOffset, u64 winOffset,
216 : HcclReduceOp opType, HcclDataType dataType)
217 : {
218 2 : const u64 startTime = KFC_GET_START_TIME();
219 2 : auto ctx = AicpuGetComContext();
220 2 : u32 rankId = ctx->rankId;
221 2 : AicpuComRankInfo *rankInfo = &ctx->rankInfo[rankId];
222 2 : void *window = reinterpret_cast<void *>(static_cast<const uintptr_t>(rankInfo->window));
223 2 : void *src = static_cast<void *>(static_cast<s8 *>(window) + winOffset + rcvOffset);
224 2 : void *dst = static_cast<void *>(static_cast<s8 *>(rcvAddr) + rcvOffset);
225 :
226 2 : CHK_RET(AicpuDispatcher::CopyData(rankId, src, dst, dataSize, dataType, opType, rankId));
227 :
228 1 : RECORD_FILL_SQE_TIME(startTime);
229 1 : return HCCL_SUCCESS;
230 : }
231 :
232 2 : HcclResult TaskOrchestrator::SelfCpySnd2WinEx1(void *sndAddr, u64 dataSize, u64 sndOffset, u64 winOffset,
233 : HcclReduceOp opType, HcclDataType dataType, u32 maxStreamNum)
234 : {
235 2 : if (maxStreamNum == 0) {
236 1 : HCCL_ERROR("max stream num can not be zero");
237 1 : return HCCL_E_PARA;
238 : }
239 1 : const u64 startTime = KFC_GET_START_TIME();
240 1 : auto ctx = AicpuGetComContext();
241 :
242 1 : AicpuComRankInfo *rankInfo = &ctx->rankInfo[ctx->rankId];
243 1 : void *src = static_cast<void *>(static_cast<s8 *>(sndAddr) + sndOffset);
244 1 : void *dst = reinterpret_cast<void *>(static_cast<const uintptr_t>(rankInfo->window) + winOffset);
245 :
246 1 : CHK_RET(AicpuDispatcher::CopyData(ctx->rankId % maxStreamNum, src, dst, dataSize, dataType, opType, ctx->rankId));
247 :
248 1 : RECORD_FILL_SQE_TIME(startTime);
249 1 : return HCCL_SUCCESS;
250 : }
251 :
252 1 : HcclResult TaskOrchestrator::SelfCpySnd2RcvEx(void *sndAddr, void *rcvAddr, u64 sndOffsets, u64 rcvOffsets,
253 : u64 dataSize, HcclReduceOp opType, HcclDataType dataType)
254 : {
255 1 : const u64 startTime = KFC_GET_START_TIME();
256 1 : auto ctx = AicpuGetComContext();
257 1 : u32 maxStreamNum = ctx->rankNum;
258 :
259 1 : void *src = static_cast<void *>(static_cast<s8 *>(sndAddr) + sndOffsets);
260 1 : void *dst = static_cast<void *>(static_cast<s8 *>(rcvAddr) + rcvOffsets);
261 1 : CHK_RET(AicpuDispatcher::CopyData(ctx->rankId % maxStreamNum, src, dst, dataSize, dataType, opType, ctx->rankId));
262 :
263 1 : RECORD_FILL_SQE_TIME(startTime);
264 1 : return HCCL_SUCCESS;
265 : }
266 :
267 36 : HcclResult TaskOrchestrator::SelfCpyWin2RcvEx(u32 mainRankId, void *rcvAddr, u64 dataSize, u64 winOffset, u64 rcvOffset,
268 : HcclReduceOp opType, HcclDataType dataType, u32 maxStreamNum)
269 : {
270 36 : if (maxStreamNum == 0) {
271 2 : HCCL_ERROR("max stream num can not be zero");
272 2 : return HCCL_E_PARA;
273 : }
274 34 : const u64 startTime = KFC_GET_START_TIME();
275 34 : auto ctx = AicpuGetComContext();
276 34 : u32 rankId = ctx->rankId;
277 34 : AicpuComRankInfo *rankInfo = &ctx->rankInfo[mainRankId];
278 :
279 34 : void *src = reinterpret_cast<void *>(static_cast<const uintptr_t>(rankInfo->window) + winOffset);
280 34 : void *dst = static_cast<void *>(static_cast<s8 *>(rcvAddr) + rcvOffset);
281 :
282 34 : CHK_RET(AicpuDispatcher::CopyData(rankId % maxStreamNum, src, dst, dataSize, dataType, opType, mainRankId));
283 :
284 34 : RECORD_FILL_SQE_TIME(startTime);
285 34 : return HCCL_SUCCESS;
286 : }
287 :
288 1 : HcclResult TaskOrchestrator::SelfCpySnd2Rcv(void *sndAddr, void *rcvAddr, u64 sndOffsets, u64 rcvOffsets, u64 dataSize,
289 : HcclReduceOp opType, HcclDataType dataType)
290 : {
291 1 : const u64 startTime = KFC_GET_START_TIME();
292 1 : auto ctx = AicpuGetComContext();
293 1 : u32 maxStreamNum = ctx->rankNum;
294 :
295 1 : void *src = static_cast<void *>(static_cast<s8 *>(sndAddr) + sndOffsets);
296 1 : void *dst = static_cast<void *>(static_cast<s8 *>(rcvAddr) + rcvOffsets);
297 1 : CHK_RET(AicpuDispatcher::CopyData(ctx->rankId % maxStreamNum, src, dst, dataSize, dataType, opType, ctx->rankId));
298 :
299 1 : RECORD_FILL_SQE_TIME(startTime);
300 1 : return HCCL_SUCCESS;
301 : }
302 :
303 8 : HcclResult TaskOrchestrator::IpcCpySnd2Win(void *sndAddr, u64 dataSize, u64 *sndOffsets, u64 *winOffsets,
304 : HcclReduceOp opType, HcclDataType dataType)
305 : {
306 8 : const u64 startTime = KFC_GET_START_TIME();
307 8 : auto ctx = AicpuGetComContext();
308 72 : for (u32 index = 0; index < ctx->rankNum; index++) {
309 64 : if (index != ctx->rankId) {
310 56 : AicpuComRankInfo *rankInfo = &ctx->rankInfo[index];
311 56 : u64 srcOffset = (sndOffsets == nullptr) ? 0 : sndOffsets[index];
312 56 : void *src = static_cast<void *>(static_cast<s8 *>(sndAddr) + srcOffset);
313 56 : u64 dstOffset = (winOffsets == nullptr) ? 0 : winOffsets[index];
314 56 : void *dst = reinterpret_cast<void *>(static_cast<const uintptr_t>(rankInfo->window) + dstOffset);
315 56 : CHK_RET(AicpuDispatcher::CopyData(index, src, dst, dataSize, dataType, opType, index));
316 : }
317 : }
318 8 : RECORD_FILL_SQE_TIME(startTime);
319 8 : return HCCL_SUCCESS;
320 : }
321 :
322 1 : HcclResult TaskOrchestrator::IpcCpySnd2Win(void *sndAddr, u64 dataSize, u64 srcOffset, u64 dstOffset,
323 : HcclReduceOp opType, HcclDataType dataType)
324 : {
325 1 : const u64 startTime = KFC_GET_START_TIME();
326 1 : auto ctx = AicpuGetComContext();
327 9 : for (u32 index = 0; index < ctx->rankNum; index++) {
328 8 : if (index != ctx->rankId) {
329 7 : AicpuComRankInfo *rankInfo = &ctx->rankInfo[index];
330 7 : void *src = static_cast<void *>(static_cast<s8 *>(sndAddr) + srcOffset);
331 7 : void *dst = reinterpret_cast<void *>(static_cast<const uintptr_t>(rankInfo->window) + dstOffset);
332 7 : CHK_RET(AicpuDispatcher::CopyData(index, src, dst, dataSize, dataType, opType, index));
333 : }
334 : }
335 1 : RECORD_FILL_SQE_TIME(startTime);
336 1 : return HCCL_SUCCESS;
337 : }
338 :
339 1 : HcclResult TaskOrchestrator::IpcCpySnd2Win(void *sndAddr, const std::vector<u64> &dataSizes,
340 : const std::vector<u64> &sndOffsets, u64 *winOffsets, HcclReduceOp opType, HcclDataType dataType)
341 : {
342 1 : const u64 startTime = KFC_GET_START_TIME();
343 1 : auto ctx = AicpuGetComContext();
344 9 : for (u32 index = 0; index < ctx->rankNum; index++) {
345 8 : if (index != ctx->rankId) {
346 7 : AicpuComRankInfo *rankInfo = &ctx->rankInfo[index];
347 7 : u64 srcOffset = sndOffsets.empty() ? 0 : sndOffsets[index];
348 7 : void *src = static_cast<void *>(static_cast<s8 *>(sndAddr) + srcOffset);
349 7 : u64 dstOffset = (winOffsets == nullptr) ? 0 : winOffsets[index];
350 7 : void *dst = reinterpret_cast<void *>(static_cast<const uintptr_t>(rankInfo->window) + dstOffset);
351 :
352 7 : CHK_RET(
353 : AicpuDispatcher::CopyData(index, src, dst, dataSizes.empty() ? 0 : dataSizes[index],
354 : dataType, opType, index));
355 : }
356 : }
357 1 : RECORD_FILL_SQE_TIME(startTime);
358 1 : return HCCL_SUCCESS;
359 : }
360 :
361 9 : HcclResult TaskOrchestrator::IpcCpySnd2Win(void *sndAddr, u64 dataSize, u64 *sndOffsets, u64 winOffsets,
362 : HcclReduceOp opType, HcclDataType dataType)
363 : {
364 9 : const u64 startTime = KFC_GET_START_TIME();
365 9 : auto ctx = AicpuGetComContext();
366 75 : for (u32 index = 0; index < ctx->rankNum; index++) {
367 66 : if (index != ctx->rankId) {
368 57 : AicpuComRankInfo *rankInfo = &ctx->rankInfo[index];
369 57 : u64 srcOffset = (sndOffsets == nullptr) ? 0 : sndOffsets[index];
370 57 : void *src = static_cast<void *>(static_cast<s8 *>(sndAddr) + srcOffset);
371 57 : void *dst = reinterpret_cast<void *>(static_cast<const uintptr_t>(rankInfo->window) + winOffsets);
372 57 : CHK_RET(AicpuDispatcher::CopyData(index, src, dst, dataSize, dataType, opType, index));
373 : }
374 : }
375 9 : RECORD_FILL_SQE_TIME(startTime);
376 9 : return HCCL_SUCCESS;
377 : }
378 :
379 2 : HcclResult TaskOrchestrator::IpcCpySnd2Win(void *sndAddr, u64 *dataSize, u64 *sndOffsets, u64 *winOffsets,
380 : HcclReduceOp opType, HcclDataType dataType)
381 : {
382 2 : const u64 startTime = KFC_GET_START_TIME();
383 2 : auto ctx = AicpuGetComContext();
384 5 : for (u32 index = 0; index < ctx->rankNum; index++) {
385 3 : if (index != ctx->rankId) {
386 1 : AicpuComRankInfo *rankInfo = &ctx->rankInfo[index];
387 1 : u64 srcOffset = (sndOffsets == nullptr) ? 0 : sndOffsets[index];
388 1 : void *src = static_cast<void *>(static_cast<s8 *>(sndAddr) + srcOffset);
389 1 : u64 dstOffset = (winOffsets == nullptr) ? 0 : winOffsets[index];
390 1 : void *dst = reinterpret_cast<void *>(static_cast<const uintptr_t>(rankInfo->window) + dstOffset);
391 1 : CHK_RET(AicpuDispatcher::CopyData(index, src, dst, dataSize[index], dataType, opType, index));
392 : }
393 : }
394 2 : RECORD_FILL_SQE_TIME(startTime);
395 2 : return HCCL_SUCCESS;
396 : }
397 :
398 : // 将本端 snd 发送至对端 window
399 1 : HcclResult TaskOrchestrator::IpcCpySnd2WinP2P(void *sndAddr, u32 dstRank, u64 dataSize, u64 sndOffsets, u64 winOffsets,
400 : HcclReduceOp opType, HcclDataType dataType)
401 : {
402 1 : const u64 startTime = KFC_GET_START_TIME();
403 1 : auto ctx = AicpuGetComContext();
404 1 : u32 selfRank = ctx->rankId;
405 :
406 1 : void *src = static_cast<void *>(static_cast<s8 *>(sndAddr) + sndOffsets);
407 1 : void *dst = reinterpret_cast<void *>(static_cast<const uintptr_t>(ctx->rankInfo[dstRank].window) + winOffsets);
408 : // 下发到主流上
409 1 : CHK_RET(AicpuDispatcher::CopyData(selfRank, src, dst, dataSize, dataType, opType, dstRank));
410 1 : RECORD_FILL_SQE_TIME(startTime);
411 1 : return HCCL_SUCCESS;
412 : }
413 :
414 : // 从对端window拷贝到本端window
415 1 : HcclResult TaskOrchestrator::IpcCpyWin2WinP2P(u32 srcRank, u64 dataSize, u64 srcOffsets, u64 dstOffsets,
416 : HcclReduceOp opType, HcclDataType dataType)
417 : {
418 1 : const u64 startTime = KFC_GET_START_TIME();
419 1 : auto ctx = AicpuGetComContext();
420 1 : u32 selfRank = ctx->rankId;
421 :
422 1 : void *src = reinterpret_cast<void *>(static_cast<const uintptr_t>(ctx->rankInfo[srcRank].window) + srcOffsets);
423 1 : void *dst = reinterpret_cast<void *>(static_cast<const uintptr_t>(ctx->rankInfo[selfRank].window) + dstOffsets);
424 1 : CHK_RET(AicpuDispatcher::CopyData(srcRank, src, dst, dataSize, dataType, opType, srcRank));
425 :
426 1 : RECORD_FILL_SQE_TIME(startTime);
427 1 : return HCCL_SUCCESS;
428 : }
429 :
430 24 : HcclResult TaskOrchestrator::IpcCpyWin2Rcv(void *rcvAddr, u64 dataSize, u64 *winOffsets, u64 *rcvOffsets,
431 : HcclReduceOp opType, HcclDataType dataType)
432 : {
433 24 : const u64 startTime = KFC_GET_START_TIME();
434 24 : auto ctx = AicpuGetComContext();
435 216 : for (u32 index = 0; index < ctx->rankNum; index++) {
436 192 : if (index != ctx->rankId) {
437 168 : AicpuComRankInfo *rankInfo = &ctx->rankInfo[index];
438 168 : u64 srcOffset = (winOffsets == nullptr) ? 0 : winOffsets[index];
439 168 : void *src = reinterpret_cast<void *>(static_cast<const uintptr_t>(rankInfo->window) + srcOffset);
440 168 : u64 dstOffset = (rcvOffsets == nullptr) ? 0 : rcvOffsets[index];
441 168 : void *dst = static_cast<void *>(static_cast<s8 *>(rcvAddr) + dstOffset);
442 :
443 168 : CHK_RET(AicpuDispatcher::CopyData(index, src, dst, dataSize, dataType, opType, index));
444 : }
445 : }
446 24 : RECORD_FILL_SQE_TIME(startTime);
447 24 : return HCCL_SUCCESS;
448 : }
449 :
450 1 : HcclResult TaskOrchestrator::IpcCpyWin2RcvEx(void *rcvAddr, u64 dataSize, u64 *rcvOffsets, u64 winOffset,
451 : HcclReduceOp opType, HcclDataType dataType)
452 : {
453 1 : const u64 startTime = KFC_GET_START_TIME();
454 1 : auto ctx = AicpuGetComContext();
455 9 : for (u32 index = 0; index < ctx->rankNum; index++) {
456 8 : if (index != ctx->rankId) {
457 7 : AicpuComRankInfo *rankInfo = &ctx->rankInfo[index];
458 7 : u64 dstOffset = (rcvOffsets == nullptr) ? 0 : rcvOffsets[index];
459 7 : void *window = reinterpret_cast<void *>(static_cast<const uintptr_t>(rankInfo->window));
460 7 : void *src = static_cast<void *>(static_cast<s8 *>(window) + winOffset + dstOffset);
461 7 : void *dst = static_cast<void *>(static_cast<s8 *>(rcvAddr) + dstOffset);
462 :
463 7 : CHK_RET(AicpuDispatcher::CopyData(index, src, dst, dataSize, dataType, opType, index));
464 : }
465 : }
466 1 : RECORD_FILL_SQE_TIME(startTime);
467 1 : return HCCL_SUCCESS;
468 : }
469 :
470 2 : HcclResult TaskOrchestrator::IpcCpyWin2RcvEx(void *rcvAddr, u64 *dataSize, u64 *rcvOffsets, u64 winOffset,
471 : HcclReduceOp opType, HcclDataType dataType)
472 : {
473 2 : const u64 startTime = KFC_GET_START_TIME();
474 2 : auto ctx = AicpuGetComContext();
475 5 : for (u32 index = 0; index < ctx->rankNum; index++) {
476 3 : if (index != ctx->rankId) {
477 1 : AicpuComRankInfo *rankInfo = &ctx->rankInfo[index];
478 1 : u64 dstOffset = (rcvOffsets == nullptr) ? 0 : rcvOffsets[index];
479 1 : void *window = reinterpret_cast<void *>(static_cast<const uintptr_t>(rankInfo->window));
480 1 : void *src = static_cast<void *>(static_cast<s8 *>(window) + winOffset + dstOffset);
481 1 : void *dst = static_cast<void *>(static_cast<s8 *>(rcvAddr) + dstOffset);
482 :
483 1 : CHK_RET(AicpuDispatcher::CopyData(index, src, dst, dataSize[index], dataType, opType, index));
484 : }
485 : }
486 2 : RECORD_FILL_SQE_TIME(startTime);
487 2 : return HCCL_SUCCESS;
488 : }
489 :
490 1 : HcclResult TaskOrchestrator::IpcCpyWin2RcvEx(void *rcvAddr, const std::vector<u64> &dataSizes,
491 : const std::vector<u64> &rcvOffsets, u64 recvOff, HcclReduceOp opType, HcclDataType dataType)
492 : {
493 1 : const u64 startTime = KFC_GET_START_TIME();
494 1 : auto ctx = AicpuGetComContext();
495 3 : for (u32 index = 0; index < ctx->rankNum; index++) {
496 2 : if (index != ctx->rankId) {
497 1 : AicpuComRankInfo *rankInfo = &ctx->rankInfo[index];
498 1 : u64 dstOffset = rcvOffsets.empty() ? 0 : rcvOffsets[index];
499 1 : void *window = reinterpret_cast<void *>(static_cast<const uintptr_t>(rankInfo->window));
500 1 : void *src = static_cast<void *>(static_cast<s8 *>(window) + recvOff + dstOffset);
501 1 : void *dst = static_cast<void *>(static_cast<s8 *>(rcvAddr) + dstOffset);
502 1 : u64 dataSize = dataSizes.empty() ? 0 : dataSizes[index];
503 1 : CHK_RET(AicpuDispatcher::CopyData(index, src, dst, dataSize, dataType, opType, index));
504 : }
505 : }
506 1 : RECORD_FILL_SQE_TIME(startTime);
507 1 : return HCCL_SUCCESS;
508 : }
509 :
510 2 : HcclResult TaskOrchestrator::IpcCpyWin2Rcv(void *rcvAddr, const std::vector<u64> &dataSizes, u64 *winOffsets,
511 : const std::vector<u64> &rcvOffsets, HcclReduceOp opType, HcclDataType dataType)
512 : {
513 2 : const u64 startTime = KFC_GET_START_TIME();
514 2 : auto ctx = AicpuGetComContext();
515 12 : for (u32 index = 0; index < ctx->rankNum; index++) {
516 10 : if (index != ctx->rankId) {
517 8 : AicpuComRankInfo *rankInfo = &ctx->rankInfo[index];
518 8 : u64 srcOffset = (winOffsets == nullptr) ? 0 : winOffsets[index];
519 8 : void *src = reinterpret_cast<void *>(static_cast<const uintptr_t>(rankInfo->window) + srcOffset);
520 8 : u64 dstOffset = rcvOffsets.empty() ? 0 : rcvOffsets[index];
521 8 : void *dst = static_cast<void *>(static_cast<s8 *>(rcvAddr) + dstOffset);
522 :
523 8 : CHK_RET(
524 : AicpuDispatcher::CopyData(index, src, dst, dataSizes.empty() ? 0 : dataSizes[index],
525 : dataType, opType, index));
526 : }
527 : }
528 2 : RECORD_FILL_SQE_TIME(startTime);
529 2 : return HCCL_SUCCESS;
530 : }
531 :
532 12 : HcclResult TaskOrchestrator::IpcCpyWin2Rcv(void *rcvAddr, u64 dataSize, u64 winOffsets, u64 *rcvOffsets,
533 : HcclReduceOp opType, HcclDataType dataType)
534 : {
535 12 : const u64 startTime = KFC_GET_START_TIME();
536 12 : auto ctx = AicpuGetComContext();
537 108 : for (u32 index = 0; index < ctx->rankNum; index++) {
538 96 : if (index != ctx->rankId) {
539 84 : AicpuComRankInfo *rankInfo = &ctx->rankInfo[index];
540 84 : void *src = reinterpret_cast<void *>(static_cast<const uintptr_t>(rankInfo->window) + winOffsets);
541 84 : u64 dstOffset = (rcvOffsets == nullptr) ? 0 : rcvOffsets[index];
542 84 : void *dst = static_cast<void *>(static_cast<s8 *>(rcvAddr) + dstOffset);
543 :
544 84 : CHK_RET(AicpuDispatcher::CopyData(index, src, dst, dataSize, dataType, opType, index));
545 : }
546 : }
547 12 : RECORD_FILL_SQE_TIME(startTime);
548 12 : return HCCL_SUCCESS;
549 : }
550 :
551 2 : HcclResult TaskOrchestrator::IpcCpyWin2Rcv(void *rcvAddr, u64 *dataSize, u64 *winOffsets, u64 *rcvOffsets,
552 : HcclReduceOp opType, HcclDataType dataType)
553 : {
554 2 : const u64 startTime = KFC_GET_START_TIME();
555 2 : auto ctx = AicpuGetComContext();
556 5 : for (u32 index = 0; index < ctx->rankNum; index++) {
557 3 : if (index != ctx->rankId) {
558 1 : AicpuComRankInfo *rankInfo = &ctx->rankInfo[index];
559 1 : u64 srcOffset = (winOffsets == nullptr) ? 0 : winOffsets[index];
560 1 : void *src = reinterpret_cast<void *>(static_cast<const uintptr_t>(rankInfo->window) + srcOffset);
561 1 : u64 dstOffset = (rcvOffsets == nullptr) ? 0 : rcvOffsets[index];
562 1 : void *dst = static_cast<void *>(static_cast<s8 *>(rcvAddr) + dstOffset);
563 :
564 1 : CHK_RET(AicpuDispatcher::CopyData(index, src, dst, dataSize[index], dataType, opType, index));
565 : }
566 : }
567 2 : RECORD_FILL_SQE_TIME(startTime);
568 2 : return HCCL_SUCCESS;
569 : }
570 :
571 1 : HcclResult TaskOrchestrator::IpcCpyWin2RcvP2P(void *rcvAddr, u32 srcRank, u64 dataSize, u64 srcOffset, u64 dstOffset,
572 : HcclReduceOp opType, HcclDataType dataType)
573 : {
574 1 : const u64 startTime = KFC_GET_START_TIME();
575 1 : auto ctx = AicpuGetComContext();
576 :
577 1 : void *src = reinterpret_cast<void *>(static_cast<const uintptr_t>(ctx->rankInfo[srcRank].window) + srcOffset);
578 1 : void *dst = static_cast<void *>(static_cast<s8 *>(rcvAddr) + dstOffset);
579 1 : CHK_RET(AicpuDispatcher::CopyData(srcRank, src, dst, dataSize, dataType, opType, srcRank));
580 :
581 1 : RECORD_FILL_SQE_TIME(startTime);
582 1 : return HCCL_SUCCESS;
583 : }
584 :
585 1 : HcclResult TaskOrchestrator::IpcCpyWin2RcvP2PMainStream(void *rcvAddr, u32 srcRank, u64 dataSize, u64 srcOffset,
586 : u64 dstOffset, HcclReduceOp opType, HcclDataType dataType)
587 : {
588 1 : const u64 startTime = KFC_GET_START_TIME();
589 1 : auto ctx = AicpuGetComContext();
590 1 : u32 selfRank = ctx->rankId;
591 :
592 1 : void *src = reinterpret_cast<void *>(static_cast<const uintptr_t>(ctx->rankInfo[srcRank].window) + srcOffset);
593 1 : void *dst = static_cast<void *>(static_cast<s8 *>(rcvAddr) + dstOffset);
594 1 : CHK_RET(AicpuDispatcher::CopyData(selfRank, src, dst, dataSize, dataType, opType, srcRank));
595 :
596 1 : RECORD_FILL_SQE_TIME(startTime);
597 1 : return HCCL_SUCCESS;
598 : }
599 :
600 3 : HcclResult TaskOrchestrator::IpcCpySnd2WinEx(void *sndAddr, u64 dataSize, u64 *sndOffsets, u64 *winOffsets,
601 : HcclReduceOp opType, HcclDataType dataType, u32 subStart, u32 subEnd, u32 maxStreamNum, bool onMainSq)
602 : {
603 3 : if (maxStreamNum == 0) {
604 2 : HCCL_ERROR("max stream num can not be zero");
605 2 : return HCCL_E_PARA;
606 : }
607 1 : const u64 startTime = KFC_GET_START_TIME();
608 1 : auto ctx = AicpuGetComContext();
609 1 : u32 streamId = 0;
610 3 : for (u32 index = subStart; index <= subEnd; index++) {
611 2 : if (index != ctx->rankId) {
612 1 : streamId = (onMainSq == true) ? (ctx->rankId % maxStreamNum) : (index % maxStreamNum);
613 1 : AicpuComRankInfo *rankInfo = &ctx->rankInfo[index];
614 1 : u64 srcOffset = (sndOffsets == nullptr) ? 0 : sndOffsets[index];
615 1 : void *src = static_cast<void *>(static_cast<s8 *>(sndAddr) + srcOffset);
616 1 : u64 dstOffset = (winOffsets == nullptr) ? 0 : winOffsets[index];
617 1 : void *dst = reinterpret_cast<void *>(static_cast<const uintptr_t>(rankInfo->window) + dstOffset);
618 :
619 1 : CHK_RET(AicpuDispatcher::CopyData(streamId, src, dst, dataSize, dataType, opType, index));
620 : }
621 : }
622 1 : RECORD_FILL_SQE_TIME(startTime);
623 1 : return HCCL_SUCCESS;
624 : }
625 :
626 4 : HcclResult TaskOrchestrator::IpcCpyWin2RcvEx(void *rcvAddr, u64 dataSize, u64 *winOffsets, u64 *rcvOffsets,
627 : HcclReduceOp opType, HcclDataType dataType, u32 subStart, u32 subEnd, u32 maxStreamNum, bool onMainSq)
628 : {
629 4 : if (maxStreamNum == 0) {
630 2 : HCCL_ERROR("max stream num can not be zero");
631 2 : return HCCL_E_PARA;
632 : }
633 2 : const u64 startTime = KFC_GET_START_TIME();
634 2 : auto ctx = AicpuGetComContext();
635 2 : u32 streamId = 0;
636 5 : for (u32 index = subStart; index <= subEnd; index++) {
637 3 : if (index != ctx->rankId) {
638 1 : streamId = (onMainSq == true) ? (ctx->rankId % maxStreamNum) : (index % maxStreamNum);
639 1 : AicpuComRankInfo *rankInfo = &ctx->rankInfo[index];
640 1 : u64 srcOffset = (winOffsets == nullptr) ? 0 : winOffsets[index];
641 1 : void *src = reinterpret_cast<void *>(static_cast<const uintptr_t>(rankInfo->window) + srcOffset);
642 1 : u64 dstOffset = (rcvOffsets == nullptr) ? 0 : rcvOffsets[index];
643 1 : void *dst = static_cast<void *>(static_cast<s8 *>(rcvAddr) + dstOffset);
644 :
645 1 : CHK_RET(AicpuDispatcher::CopyData(streamId, src, dst, dataSize, dataType, opType, index));
646 : }
647 : }
648 2 : RECORD_FILL_SQE_TIME(startTime);
649 2 : return HCCL_SUCCESS;
650 : }
651 :
652 1 : HcclResult TaskOrchestrator::IpcCpySnd2WinSliceEx(void *sndAddr, std::vector<Slice> &dataSlice, u64 *winOffsets,
653 : HcclReduceOp opType, HcclDataType dataType, u32 subStart, u32 subEnd, u32 maxStreamNum, bool onMainSq)
654 : {
655 1 : if (maxStreamNum == 0) {
656 1 : HCCL_ERROR("max stream num can not be zero");
657 1 : return HCCL_E_PARA;
658 : }
659 0 : const u64 startTime = KFC_GET_START_TIME();
660 0 : auto ctx = AicpuGetComContext();
661 0 : u32 streamId = 0;
662 0 : for (u32 index = subStart; index <= subEnd; index++) {
663 0 : if (index != ctx->rankId) {
664 0 : streamId = (onMainSq == true) ? (ctx->rankId % maxStreamNum) : (index % maxStreamNum);
665 0 : AicpuComRankInfo *rankInfo = &ctx->rankInfo[index];
666 0 : u64 srcOffset = dataSlice[index].offset;
667 0 : void *src = static_cast<void *>(static_cast<s8 *>(sndAddr) + srcOffset);
668 0 : u64 dstOffset = (winOffsets == nullptr) ? 0 : winOffsets[index];
669 0 : void *dst = reinterpret_cast<void *>(static_cast<const uintptr_t>(rankInfo->window) + dstOffset);
670 :
671 0 : CHK_RET(AicpuDispatcher::CopyData(streamId, src, dst, dataSlice[index].size, dataType, opType, index));
672 : }
673 : }
674 0 : RECORD_FILL_SQE_TIME(startTime);
675 0 : return HCCL_SUCCESS;
676 : }
677 :
678 2 : HcclResult TaskOrchestrator::IpcCpyWin2RcvSliceEx(void *rcvAddr, std::vector<Slice> &dataSlice, u64 *winOffsets,
679 : HcclReduceOp opType, HcclDataType dataType, u32 subStart, u32 subEnd, u32 maxStreamNum, bool onMainSq)
680 : {
681 2 : if (maxStreamNum == 0) {
682 1 : HCCL_ERROR("max stream num can not be zero");
683 1 : return HCCL_E_PARA;
684 : }
685 1 : const u64 startTime = KFC_GET_START_TIME();
686 1 : auto ctx = AicpuGetComContext();
687 1 : u32 streamId = 0;
688 3 : for (u32 index = subStart; index <= subEnd; index++) {
689 2 : if (index != ctx->rankId) {
690 1 : streamId = (onMainSq == true) ? (ctx->rankId % maxStreamNum) : (index % maxStreamNum);
691 1 : AicpuComRankInfo *rankInfo = &ctx->rankInfo[index];
692 1 : u64 srcOffset = (winOffsets == nullptr) ? 0 : winOffsets[index];
693 1 : void *src = reinterpret_cast<void *>(static_cast<const uintptr_t>(rankInfo->window) + srcOffset);
694 1 : u64 dstOffset = dataSlice[index].offset;
695 1 : void *dst = static_cast<void *>(static_cast<s8 *>(rcvAddr) + dstOffset);
696 :
697 1 : CHK_RET(AicpuDispatcher::CopyData(streamId, src, dst, dataSlice[index].size, dataType, opType, index));
698 : }
699 : }
700 1 : RECORD_FILL_SQE_TIME(startTime);
701 1 : return HCCL_SUCCESS;
702 : }
703 :
704 8 : HcclResult TaskOrchestrator::SelfLocalReduce(u64 dataSize, HcclReduceOp opType, HcclDataType dataType)
705 : {
706 8 : const u64 startTime = KFC_GET_START_TIME();
707 8 : auto ctx = AicpuGetComContext();
708 8 : u32 rankId = ctx->rankId;
709 8 : AicpuComRankInfo *rankInfo = &ctx->rankInfo[rankId];
710 8 : void *src = nullptr;
711 8 : void *dst = reinterpret_cast<void *>(static_cast<const uintptr_t>(rankInfo->window));
712 8 : u64 srcOffset = 0LU;
713 8 : u64 cpySize = 0LU;
714 :
715 8 : u32 rankNum = ctx->rankNum;
716 8 : u32 power = static_cast<u32>(log2(rankNum));
717 8 : u32 rankPower = static_cast<u32>(pow(2, power));
718 8 : if (rankPower < rankNum) {
719 0 : srcOffset = rankPower * dataSize;
720 0 : cpySize = (rankNum - rankPower) * dataSize;
721 0 : HCCL_DEBUG("SelfLocalReduce: rankNum %u, power %u, rankPower %u, srcOffset %lu, cpySize %lu", rankNum, power,
722 : rankPower, srcOffset, cpySize);
723 0 : src = reinterpret_cast<void *>(static_cast<const uintptr_t>(rankInfo->window) + srcOffset);
724 0 : CHK_RET(AicpuDispatcher::CopyData(rankId, src, dst, cpySize, dataType, opType, rankId));
725 : }
726 :
727 32 : for (u32 round = 0u; round < power; round++) {
728 24 : u32 sliceNum = rankPower / static_cast<u32>(pow(2, round + 1));
729 24 : srcOffset = sliceNum * dataSize;
730 24 : cpySize = srcOffset;
731 24 : HCCL_DEBUG("SelfLocalReduce: sliceNum %u, rankNum %u, power %u, rankPower %u, srcOffset %lu", sliceNum, rankNum,
732 : power, rankPower, srcOffset);
733 24 : src = reinterpret_cast<void *>(static_cast<const uintptr_t>(rankInfo->window) + srcOffset);
734 24 : CHK_RET(AicpuDispatcher::CopyData(rankId, src, dst, cpySize, dataType, opType, rankId));
735 : }
736 8 : RECORD_FILL_SQE_TIME(startTime);
737 8 : return HCCL_SUCCESS;
738 : }
739 :
740 65 : HcclResult TaskOrchestrator::LaunchTasks()
741 : {
742 65 : auto ctx = AicpuGetComContext();
743 65 : return LaunchTasksEx(0, ctx->rankNum - 1, ctx->rankNum);
744 : }
745 :
746 103 : HcclResult TaskOrchestrator::LaunchTasksEx(u32 subStart, u32 subEnd, u32 maxStreamNum)
747 : {
748 103 : if (maxStreamNum == 0) {
749 2 : HCCL_ERROR("max stream num can not be zero");
750 2 : return HCCL_E_PARA;
751 : }
752 101 : const u64 startTime = GetCurCpuTimestamp();
753 101 : auto ctx = AicpuGetComContext();
754 101 : if (AicpuKfcUtils::NeedRecordTimeTaken(*ctx)) {
755 10 : RECORD_PROF_TIME(sendTaskStartTime);
756 : }
757 101 : u32 activeRank = ctx->rankId % maxStreamNum;
758 :
759 : /* 两阶段模式，主流待正式执行时再下 */
760 : /* 一阶段第一次，可以先下主流 */
761 101 : if (ctx->directlySendMainSteramSqe) {
762 96 : CHK_PRT_RET(ActiveRecordMain(activeRank) != HCCL_SUCCESS,
763 : HCCL_ERROR("launch task failed, sqid:%u", activeRank),
764 : HCCL_E_INTERNAL);
765 : }
766 :
767 101 : auto profInst = AicpuKfcProf::GetProInst(*ctx);
768 874 : for (u32 index = subStart; index <= subEnd; index++) {
769 773 : if (index != activeRank) {
770 672 : if (AicpuKfcUtils::NeedRecordTimeTaken(*ctx)) {
771 70 : profInst.fillSqeCnt += GetSqeContext()->buffPtr[index].sqeCnt;
772 : }
773 672 : CHK_PRT_RET(AicpuDispatcher::LaunchTask(index) != HCCL_SUCCESS,
774 : HCCL_ERROR("launch task failed, sqid:%u", index), HCCL_E_INTERNAL);
775 : }
776 : }
777 101 : HCCL_INFO("LaunchTasksEx sqeBufferLocal, subStart=%u, subEnd=%u", subStart, subEnd);
778 :
779 101 : if (AicpuKfcUtils::NeedRecordTimeTaken(*ctx)) {
780 10 : const u64 endTime = GetCurCpuTimestamp();
781 10 : profInst.sendSqeTimes += endTime - startTime;
782 10 : profInst.sendSqeBatch += ctx->rankNum;
783 10 : RECORD_PROF_TIME(sendSqeFinishTime);
784 : }
785 101 : return HCCL_SUCCESS;
786 : }
787 :
788 : // 主流notify从流 从流wait主流
789 178 : HcclResult TaskOrchestrator::MainSubPreSync()
790 : {
791 178 : auto ctx = AicpuGetComContext();
792 178 : return MainSubPreSync(ctx->rankId, 0U, ctx->rankNum - 1U, ctx->rankNum);
793 : }
794 :
795 0 : HcclResult TaskOrchestrator::MainSubPreSync(const uint32_t subStream)
796 : {
797 0 : auto ctx = AicpuGetComContext();
798 0 : return MainSubPreSync(ctx->rankId, subStream, subStream, ctx->rankNum);
799 : }
800 :
801 179 : HcclResult TaskOrchestrator::MainSubPreSync(uint32_t mainStream, uint32_t subStart, uint32_t subEnd, uint32_t maxStream)
802 : {
803 179 : if (maxStream == 0U) {
804 1 : HCCL_ERROR("Max stream num can not be zero");
805 1 : return HCCL_E_PARA;
806 : }
807 178 : const u64 startTime = KFC_GET_START_TIME();
808 1532 : for (u32 index = subStart; index <= subEnd; index++) {
809 1354 : if (index != mainStream) {
810 1176 : CHK_RET(AicpuDispatcher::SignalRecord(mainStream % maxStream, index, AicpuDispatcher::NO_IPC,
811 : AicpuDispatcher::PRE_SYNC));
812 1176 : CHK_RET(AicpuDispatcher::SignalWait(index % maxStream, index, AicpuDispatcher::NO_IPC,
813 : AicpuDispatcher::PRE_SYNC));
814 : }
815 : }
816 178 : RECORD_FILL_SQE_TIME(startTime);
817 178 : return HCCL_SUCCESS;
818 : }
819 :
820 : // 从流notify主流 主流wait从流
821 110 : HcclResult TaskOrchestrator::MainSubPostSync()
822 : {
823 110 : auto ctx = AicpuGetComContext();
824 110 : return MainSubPostSync(ctx->rankId, 0U, ctx->rankNum - 1U, ctx->rankNum);
825 : }
826 :
827 0 : HcclResult TaskOrchestrator::MainSubPostSync(const uint32_t subStream)
828 : {
829 0 : auto ctx = AicpuGetComContext();
830 0 : return MainSubPostSync(ctx->rankId, subStream, subStream, ctx->rankNum);
831 : }
832 :
833 111 : HcclResult TaskOrchestrator::MainSubPostSync(uint32_t mainStream, uint32_t subStart, uint32_t subEnd,
834 : uint32_t maxStream)
835 : {
836 111 : if (maxStream == 0U) {
837 1 : HCCL_ERROR("Max stream num can not be zero");
838 1 : return HCCL_E_PARA;
839 : }
840 110 : const u64 startTime = KFC_GET_START_TIME();
841 934 : for (uint32_t index = subStart; index <= subEnd; index++) {
842 824 : if (index != mainStream) {
843 714 : CHK_RET(AicpuDispatcher::SignalRecord(index % maxStream, index, AicpuDispatcher::NO_IPC,
844 : AicpuDispatcher::POST_SYNC));
845 714 : CHK_RET(AicpuDispatcher::SignalWait(mainStream % maxStream, index, AicpuDispatcher::NO_IPC,
846 : AicpuDispatcher::POST_SYNC));
847 : }
848 : }
849 110 : RECORD_FILL_SQE_TIME(startTime);
850 110 : return HCCL_SUCCESS;
851 : }
852 :
853 55 : HcclResult TaskOrchestrator::IpcPreSync()
854 : {
855 55 : const u64 startTime = KFC_GET_START_TIME();
856 55 : auto ctx = AicpuGetComContext();
857 467 : for (u32 index = 0; index < ctx->rankNum; index++) {
858 412 : if (index != ctx->rankId) {
859 357 : CHK_RET(AicpuDispatcher::SignalRecord(index, index, AicpuDispatcher::IPC, AicpuDispatcher::PRE_SYNC));
860 357 : CHK_RET(AicpuDispatcher::SignalWait(index, index, AicpuDispatcher::IPC, AicpuDispatcher::PRE_SYNC));
861 : }
862 : }
863 55 : RECORD_FILL_SQE_TIME(startTime);
864 55 : return HCCL_SUCCESS;
865 : }
866 :
867 69 : HcclResult TaskOrchestrator::IpcPreRecordEx(u32 subStart, u32 subEnd, u32 maxStreamNum, bool onMainSq)
868 : {
869 69 : if (maxStreamNum == 0) {
870 1 : HCCL_ERROR("max stream num can not be zero");
871 1 : return HCCL_E_PARA;
872 : }
873 68 : const u64 startTime = KFC_GET_START_TIME();
874 68 : u32 stream_id = 0;
875 68 : auto ctx = AicpuGetComContext();
876 598 : for (u32 index = subStart; index <= subEnd; index++) {
877 530 : if (index != ctx->rankId) {
878 462 : stream_id = (onMainSq == true) ? (ctx->rankId % maxStreamNum) : (index % maxStreamNum);
879 462 : CHK_RET(AicpuDispatcher::SignalRecord(stream_id, index, AicpuDispatcher::IPC, AicpuDispatcher::PRE_SYNC));
880 : }
881 : }
882 :
883 68 : RECORD_FILL_SQE_TIME(startTime);
884 68 : return HCCL_SUCCESS;
885 : }
886 :
887 1 : HcclResult TaskOrchestrator::IpcPreWaitEx(u32 subStart, u32 subEnd, u32 maxStreamNum, bool onMainSq)
888 : {
889 1 : if (maxStreamNum == 0) {
890 1 : HCCL_ERROR("max stream num can not be zero");
891 1 : return HCCL_E_PARA;
892 : }
893 0 : const u64 startTime = KFC_GET_START_TIME();
894 0 : u32 stream_id = 0;
895 0 : auto ctx = AicpuGetComContext();
896 0 : for (u32 index = subStart; index <= subEnd; index++) {
897 0 : if (index != ctx->rankId) {
898 0 : stream_id = (onMainSq == true) ? (ctx->rankId % maxStreamNum) : (index % maxStreamNum);
899 0 : CHK_RET(AicpuDispatcher::SignalWait(stream_id, index, AicpuDispatcher::IPC, AicpuDispatcher::PRE_SYNC));
900 : }
901 : }
902 :
903 0 : RECORD_FILL_SQE_TIME(startTime);
904 0 : return HCCL_SUCCESS;
905 : }
906 :
907 5 : HcclResult TaskOrchestrator::IpcPreSyncEx(u32 subStart, u32 subEnd, u32 maxStreamNum, bool onMainSq)
908 : {
909 5 : if (maxStreamNum == 0) {
910 3 : HCCL_ERROR("max stream num can not be zero");
911 3 : return HCCL_E_PARA;
912 : }
913 2 : const u64 startTime = KFC_GET_START_TIME();
914 2 : u32 stream_id = 0;
915 2 : auto ctx = AicpuGetComContext();
916 4 : for (u32 index = subStart; index <= subEnd; index++) {
917 2 : if (index != ctx->rankId) {
918 1 : stream_id = (onMainSq == true) ? (ctx->rankId % maxStreamNum) : (index % maxStreamNum);
919 1 : CHK_RET(AicpuDispatcher::SignalRecord(stream_id, index, AicpuDispatcher::IPC, AicpuDispatcher::PRE_SYNC));
920 1 : CHK_RET(AicpuDispatcher::SignalWait(stream_id, index, AicpuDispatcher::IPC, AicpuDispatcher::PRE_SYNC));
921 : }
922 : }
923 :
924 2 : RECORD_FILL_SQE_TIME(startTime);
925 2 : return HCCL_SUCCESS;
926 : }
927 :
928 1 : HcclResult TaskOrchestrator::IpcPreSyncOnMainStream()
929 : {
930 1 : const u64 startTime = KFC_GET_START_TIME();
931 1 : auto ctx = AicpuGetComContext();
932 9 : for (u32 index = 0; index < ctx->rankNum; index++) {
933 8 : if (index != ctx->rankId) {
934 7 : CHK_RET(AicpuDispatcher::SignalRecord(ctx->rankId, index, AicpuDispatcher::IPC, AicpuDispatcher::PRE_SYNC));
935 7 : CHK_RET(AicpuDispatcher::SignalWait(ctx->rankId, index, AicpuDispatcher::IPC, AicpuDispatcher::PRE_SYNC));
936 : }
937 : }
938 1 : RECORD_FILL_SQE_TIME(startTime);
939 1 : return HCCL_SUCCESS;
940 : }
941 :
942 0 : HcclResult TaskOrchestrator::IpcPostSyncOnMainStream()
943 : {
944 0 : const u64 startTime = KFC_GET_START_TIME();
945 0 : auto ctx = AicpuGetComContext();
946 0 : for (u32 index = 0; index < ctx->rankNum; index++) {
947 0 : if (index != ctx->rankId) {
948 0 : CHK_RET(
949 : AicpuDispatcher::SignalRecord(ctx->rankId, index, AicpuDispatcher::IPC, AicpuDispatcher::POST_SYNC));
950 0 : CHK_RET(AicpuDispatcher::SignalWait(ctx->rankId, index, AicpuDispatcher::IPC, AicpuDispatcher::POST_SYNC));
951 : }
952 : }
953 0 : RECORD_FILL_SQE_TIME(startTime);
954 0 : return HCCL_SUCCESS;
955 : }
956 :
957 55 : HcclResult TaskOrchestrator::IpcPostSync()
958 : {
959 55 : const u64 startTime = KFC_GET_START_TIME();
960 55 : auto ctx = AicpuGetComContext();
961 467 : for (u32 index = 0; index < ctx->rankNum; index++) {
962 412 : if (index != ctx->rankId) {
963 357 : CHK_RET(AicpuDispatcher::SignalRecord(index, index, AicpuDispatcher::IPC, AicpuDispatcher::POST_SYNC));
964 357 : CHK_RET(AicpuDispatcher::SignalWait(index, index, AicpuDispatcher::IPC, AicpuDispatcher::POST_SYNC));
965 : }
966 : }
967 55 : RECORD_FILL_SQE_TIME(startTime);
968 55 : return HCCL_SUCCESS;
969 : }
970 :
971 1 : HcclResult TaskOrchestrator::IpcPostRecordEx(u32 subStart, u32 subEnd, u32 maxStreamNum, bool onMainSq)
972 : {
973 1 : if (maxStreamNum == 0) {
974 1 : HCCL_ERROR("max stream num can not be zero");
975 1 : return HCCL_E_PARA;
976 : }
977 0 : const u64 startTime = KFC_GET_START_TIME();
978 0 : u32 stream_id = 0;
979 0 : auto ctx = AicpuGetComContext();
980 0 : for (u32 index = subStart; index <= subEnd; index++) {
981 0 : if (index != ctx->rankId) {
982 0 : stream_id = (onMainSq == true) ? (ctx->rankId % maxStreamNum) : (index % maxStreamNum);
983 0 : CHK_RET(AicpuDispatcher::SignalRecord(stream_id, index, AicpuDispatcher::IPC, AicpuDispatcher::POST_SYNC));
984 : }
985 : }
986 :
987 0 : RECORD_FILL_SQE_TIME(startTime);
988 0 : return HCCL_SUCCESS;
989 : }
990 :
991 69 : HcclResult TaskOrchestrator::IpcPostWaitEx(u32 subStart, u32 subEnd, u32 maxStreamNum, bool onMainSq)
992 : {
993 69 : if (maxStreamNum == 0) {
994 1 : HCCL_ERROR("max stream num can not be zero");
995 1 : return HCCL_E_PARA;
996 : }
997 68 : const u64 startTime = KFC_GET_START_TIME();
998 68 : u32 stream_id = 0;
999 68 : auto ctx = AicpuGetComContext();
1000 598 : for (u32 index = subStart; index <= subEnd; index++) {
1001 530 : if (index != ctx->rankId) {
1002 462 : stream_id = (onMainSq == true) ? (ctx->rankId % maxStreamNum) : (index % maxStreamNum);
1003 462 : CHK_RET(AicpuDispatcher::SignalWait(stream_id, index, AicpuDispatcher::IPC, AicpuDispatcher::POST_SYNC));
1004 : }
1005 : }
1006 :
1007 68 : RECORD_FILL_SQE_TIME(startTime);
1008 68 : return HCCL_SUCCESS;
1009 : }
1010 :
1011 5 : HcclResult TaskOrchestrator::IpcPostSyncEx(u32 subStart, u32 subEnd, u32 maxStreamNum, bool onMainSq)
1012 : {
1013 5 : if (maxStreamNum == 0) {
1014 3 : HCCL_ERROR("max stream num can not be zero");
1015 3 : return HCCL_E_PARA;
1016 : }
1017 2 : const u64 startTime = KFC_GET_START_TIME();
1018 2 : u32 stream_id = 0;
1019 2 : auto ctx = AicpuGetComContext();
1020 4 : for (u32 index = subStart; index <= subEnd; index++) {
1021 2 : if (index != ctx->rankId) {
1022 1 : stream_id = (onMainSq == true) ? (ctx->rankId % maxStreamNum) : (index % maxStreamNum);
1023 1 : CHK_RET(AicpuDispatcher::SignalRecord(stream_id, index, AicpuDispatcher::IPC, AicpuDispatcher::POST_SYNC));
1024 1 : CHK_RET(AicpuDispatcher::SignalWait(stream_id, index, AicpuDispatcher::IPC, AicpuDispatcher::POST_SYNC));
1025 : }
1026 : }
1027 :
1028 2 : RECORD_FILL_SQE_TIME(startTime);
1029 2 : return HCCL_SUCCESS;
1030 : }
1031 :
1032 203 : HcclResult TaskOrchestrator::ActiveRecordMain(u16 sqId)
1033 : {
1034 203 : auto ctx = AicpuGetComContext();
1035 203 : HcclComStreamInfo *streamInfo = &ctx->streamInfo[sqId];
1036 203 : HCCL_DEBUG("ActiveStream rankId:%d, devId:%d, sqId:%lu, sqeCnt:%d",
1037 : sqId,
1038 : ctx->devId,
1039 : streamInfo->sqId,
1040 : GetSqeContext()->buffPtr[sqId].sqeCnt);
1041 203 : if (GetSqeContext()->buffPtr[sqId].sqeCnt == 0U) {
1042 27 : return HCCL_SUCCESS;
1043 : }
1044 176 : if (AicpuKfcUtils::NeedRecordTimeTaken(*ctx)) {
1045 15 : AicpuKfcProf::GetProInst(*ctx).fillSqeCnt += GetSqeContext()->buffPtr[sqId].sqeCnt;
1046 : }
1047 176 : CHK_PRT_RET(AicpuDispatcher::LaunchTask(sqId) != HCCL_SUCCESS,
1048 : HCCL_ERROR("Launch task failed, sqid:%u", sqId),
1049 : HCCL_E_INTERNAL);
1050 176 : return HCCL_SUCCESS;
1051 : }
1052 :
1053 51 : HcclResult TaskOrchestrator::WaitMainStreamFinish(AicpuComContext *ctx)
1054 : {
1055 51 : s32 sqId = ctx->streamInfo[ctx->rankId].sqId;
1056 51 : HCCL_INFO("Start WaitMainStreamFinish..devId = %d rankId:%u, sqid:%d", ctx->devId, ctx->rankId, sqId);
1057 :
1058 51 : auto ret = WaitFinishWhileLoop(ctx);
1059 51 : if (ret != HCCL_SUCCESS) {
1060 13 : if (ret != HCCL_E_SUSPENDING) {
1061 2 : HCCL_ERROR("WaitFinishWhileLoop failed, determinism %u, ret %u.", ctx->determinism, ret);
1062 : }
1063 13 : return ret;
1064 : }
1065 38 : HCCL_INFO("End WaitMainStreamFinish..devId = %d rankid:%u, sqid:%d", ctx->devId, ctx->rankId, sqId);
1066 :
1067 38 : return HCCL_SUCCESS;
1068 : }
1069 :
1070 462727 : bool TaskOrchestrator::IsTaskExceptionForHccs(AicpuComContext *ctx)
1071 : {
1072 462727 : if (ctx->dfxExtendInfo.cqeStatus != dfx::CqeStatus::kCqeException) {
1073 462727 : return false;
1074 : }
1075 :
1076 : // NOTE: 需要task exception补全dfx能力，定位故障task的remote rank; 目前暂不具备识别是否跨片的能力，默认失败的task均为跨片操作。
1077 0 : if (ctx->dfxExtendInfo.cqeException.sqeType == RT_STARS_SQE_TYPE_WRITE_VALUE ||
1078 0 : ctx->dfxExtendInfo.cqeException.sqeType == RT_STARS_SQE_TYPE_SDMA) {
1079 0 : return true;
1080 : }
1081 0 : return false;
1082 : }
1083 :
1084 37 : HcclResult TaskOrchestrator::DealKfcCommand(AicpuComContext *ctx)
1085 : {
1086 37 : KfcCommand cmd = KfcCommand::kNone;
1087 37 : CHK_RET(AicpuHdcUtils::GetOpExecCtrlCmd(ctx->kfcControlTransferH2D, cmd));
1088 37 : if (cmd == KfcCommand::kStopLaunch) {
1089 1 : HCCL_WARNING("hccl aicpu stop wait finish, for recv stop launch cmd");
1090 1 : return HCCL_E_SUSPENDING;
1091 36 : } else if ((cmd == KfcCommand::NsStopLaunch) && (ctx->commOpenStatus == true) && (ctx->endStopLaunch == false)) {
1092 0 : HCCL_WARNING("N second stop Launch for recv stop launch cmd.");
1093 0 : AicpuUpdatComContextMumber(offsetof(AicpuComContext, isStopLaunch), true);
1094 0 : AicpuUpdatComContextMumber(offsetof(AicpuComContext, endStopLaunch), true);
1095 0 : return HCCL_E_SUSPENDING;
1096 36 : } else if (cmd == KfcCommand::kDestroyComm) {
1097 0 : HCCL_WARNING("hccl aicpu stop wait finish, for recv destroy comm cmd");
1098 0 : return HCCL_E_SUSPENDING;
1099 36 : } else if (cmd == KfcCommand::kExit) {
1100 0 : HCCL_ERROR("hccl aicpu stop wait finish, for recv exit cmd.");
1101 0 : return HCCL_E_INTERNAL;
1102 : }
1103 :
1104 36 : return HCCL_SUCCESS;
1105 : }
1106 :
1107 37 : HcclResult TaskOrchestrator::WaitFinishWhileLoop(AicpuComContext *ctx)
1108 : {
1109 : static uint32_t logHead = UINT32_MAX;
1110 : static uint32_t logTail = UINT32_MAX;
1111 37 : const uint64_t startUsec = GetCurCpuTimestamp();
1112 :
1113 37 : int32_t sqId = ctx->streamInfo[ctx->rankId].sqId;
1114 37 : uint32_t sqHead = 0;
1115 37 : uint32_t sqTail = 0;
1116 37 : CHK_RET(QuerySqStatusByType(ctx->devId, sqId, DRV_SQCQ_PROP_SQ_TAIL, sqTail));
1117 37 : uint32_t loopCnt = 0;
1118 37 : ctx->sendCntRecord[1] = AicpuKfcUtils::GetSendCnt(ctx); // 1 记录下发完任务后的sendCnt
1119 37 : ctx->recvCntRecord[1] = AicpuKfcUtils::GetRecvCnt(ctx); // 1 记录下发完任务后的recvCnt
1120 : do {
1121 37 : if (ctx->dfxExtendInfo.pollStatus == PollStatus::kStopAsException) {
1122 0 : if (IsTaskExceptionForHccs(ctx)) {
1123 0 : HCCL_WARNING("hccl aicpu stop wait task exec finish, for task exception.");
1124 0 : return HCCL_E_SUSPENDING;
1125 : } else {
1126 0 : HCCL_ERROR("hccl aicpu exec failed, for task exception.");
1127 0 : return HCCL_E_INTERNAL;
1128 : }
1129 : }
1130 :
1131 37 : CHK_RET(DealKfcCommand(ctx));
1132 36 : CHK_RET(QuerySqStatusByType(ctx->devId, sqId, DRV_SQCQ_PROP_SQ_HEAD, sqHead));
1133 36 : if (loopCnt > 10000) { // 10000 is max loop cnt
1134 0 : uint32_t overflowFlag = 0;
1135 0 : OverflowAddrCheck(ctx, overflowFlag, sqHead, sqTail);
1136 0 : loopCnt = 0;
1137 0 : if (logHead != sqHead || logTail != sqTail) {
1138 0 : logHead = sqHead;
1139 0 : logTail = sqTail;
1140 0 : HCCL_INFO("Current state. devId:%u sqid:%d, head:%u, tail:%u", ctx->devId, sqId, sqHead, sqTail);
1141 : }
1142 0 : CHK_RET(WorkSpacePrint(ctx));
1143 : }
1144 36 : CHK_RET(CheckTaskTimeout(ctx, startUsec));
1145 36 : HCCL_INFO("Current state. loopCnt:%u, devId:%u sqid:%d, head:%u, tail:%u", loopCnt, ctx->devId, sqId, sqHead, sqTail);
1146 36 : loopCnt++;
1147 36 : } while (sqHead != sqTail);
1148 36 : return HCCL_SUCCESS;
1149 : }
1150 :
1151 9 : void TaskOrchestrator::PrintTimeOutSqInfo(AicpuComContext *ctx, u64 timeThreshold)
1152 : {
1153 9 : uint32_t status = 0U;
1154 9 : int32_t sqId = ctx->streamInfo[ctx->rankId].sqId;
1155 9 : auto ret = QuerySqStatusByType(ctx->devId, sqId, DRV_SQCQ_PROP_SQ_CQE_STATUS, status);
1156 9 : if (ret != 0) {
1157 0 : HCCL_ERROR("QuerySqStatusByType status failed. ret = %u sqid:%d", ret, sqId);
1158 : }
1159 75 : for (uint32_t i = 0U; i < ctx->rankNum; i++) {
1160 66 : uint32_t sqHead = 0U;
1161 66 : uint32_t sqTail = 0U;
1162 66 : (void)QuerySqStatus(ctx->devId, ctx->streamInfo[i].sqId, sqHead, sqTail);
1163 66 : SqeInfo sqeInfo;
1164 66 : auto headRet = AicpuSqeContext::QuerySqeInfoByHead(i, sqHead, &sqeInfo);
1165 66 : if (headRet != HCCL_SUCCESS) {
1166 50 : HCCL_ERROR("QuerySqeInfoByHead status failed. ret = %u sqHead:%d", headRet, sqHead);
1167 50 : continue;
1168 : }
1169 16 : HCCL_ERROR("KFC timeout..[%lu]s, commId %s, stream %u sqid %d head %u tail %u. SqeInfo:%s",
1170 : timeThreshold, ctx->hcomId, i, ctx->streamInfo[i].sqId, sqHead, sqTail,
1171 : AicpuSqeContext::GetString(sqeInfo).c_str());
1172 : }
1173 9 : }
1174 :
1175 37 : HcclResult TaskOrchestrator::CheckTaskTimeout(AicpuComContext *ctx, uint64_t startUsec)
1176 : {
1177 37 : const uint64_t sqeTimeoutSec = ctx->dfxExtendInfo.dfxTimeOutConfig.sqeWaitTimeOut;
1178 37 : if (GetCurCpuTimestamp() - startUsec > static_cast<uint64_t>(NSEC_PER_SEC) * sqeTimeoutSec ) {
1179 1 : PrintTimeOutSqInfo(ctx, sqeTimeoutSec);
1180 1 : CHK_RET(MC2TraceUtils::Save());
1181 1 : AicpuUpdatComContextMumber(offsetof(AicpuComContext, dfxExtendInfo.kfcStatus), DfxKfcStatus::kTimeOut);
1182 1 : return HCCL_E_TIMEOUT;
1183 : }
1184 36 : return HCCL_SUCCESS;
1185 : }
1186 :
1187 0 : HcclResult TaskOrchestrator::WorkSpacePrint(AicpuComContext *ctx)
1188 : {
1189 : static int staticSndCnt = -1;
1190 0 : uint64_t waitAddr = ctx->workSpaceAddr + ctx->notifyOff;
1191 0 : int sndCnt = static_cast<int>((reinterpret_cast<AivAicpuOpParam *>(waitAddr))->sendCnt);
1192 0 : if (staticSndCnt != sndCnt) {
1193 0 : staticSndCnt = sndCnt;
1194 0 : std::stringstream recordLog;
1195 0 : recordLog << "waitAddr:0x" << std::hex << waitAddr << ", sendCnt:" << std::dec << sndCnt;
1196 0 : HCCL_INFO("%s", recordLog.str().c_str());
1197 0 : CHK_RET(MC2TraceUtils::Submit(recordLog.str().c_str()));
1198 0 : }
1199 :
1200 : static int staticRcvCnt = -1;
1201 0 : uint64_t recordAddr = ctx->workSpaceAddr + ctx->notifyOff + ctx->notifyBeginCnt * sizeof(uint8_t) * AC_SQE_SIZE;
1202 0 : int rcvCnt = static_cast<int>((reinterpret_cast<AivAicpuOpParam *>(recordAddr))->rcvCnt);
1203 0 : if (staticRcvCnt != rcvCnt) {
1204 0 : staticRcvCnt = rcvCnt;
1205 0 : std::stringstream recordLog;
1206 0 : recordLog << "recordAddr:0x" << std::hex << recordAddr << ", rcvCnt:" << std::dec << rcvCnt;
1207 0 : HCCL_INFO("%s", recordLog.str().c_str());
1208 0 : CHK_RET(MC2TraceUtils::Submit(recordLog.str().c_str()));
1209 0 : }
1210 0 : return HCCL_SUCCESS;
1211 : }
1212 :
1213 1 : void TaskOrchestrator::OverflowAddrCheck(AicpuComContext *ctx, uint32_t &overflowFlag, uint32_t sqHead, uint32_t sqTail)
1214 : {
1215 1 : if (ctx->devType != DevType::DEV_TYPE_310P1 && ctx->devType != DevType::DEV_TYPE_310P3) {
1216 0 : return;
1217 : }
1218 :
1219 1 : if (ctx->overflowAddr == 0) {
1220 0 : return;
1221 : }
1222 :
1223 1 : uint32_t overflowValTmp = *reinterpret_cast<uint32_t *>(ctx->overflowAddr);
1224 1 : if ((overflowFlag == 0) && ((overflowValTmp & 0x11) == 0x11)) { // 与runtime对齐，溢出时会给该地址里填写0x11
1225 1 : HCCL_WARNING("data is overflow, sqHead cur head:%u tail:%u, overflowVal:%u overflowValTmp:%u", sqHead, sqTail,
1226 : overflowFlag, overflowValTmp);
1227 1 : overflowFlag = 1;
1228 : }
1229 : }
1230 :
1231 38 : HcclResult TaskOrchestrator::AddBarrier(uint32_t mainStream, uint32_t rankId, uint32_t rankNum)
1232 : {
1233 38 : const uint32_t preRankId = (rankId + rankNum - 1U) % rankNum;
1234 38 : const uint32_t postRankId = (rankId + 1U) % rankNum;
1235 : // 片间同步 notify后卡 wait前卡
1236 38 : auto ret = AicpuDispatcher::SignalRecord(mainStream, postRankId, AicpuDispatcher::IPC, AicpuDispatcher::PRE_SYNC);
1237 38 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("Add notify post rank failed"), ret);
1238 38 : ret = AicpuDispatcher::SignalWait(mainStream, preRankId, AicpuDispatcher::IPC, AicpuDispatcher::PRE_SYNC);
1239 38 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("Add wait pre rank failed"), ret);
1240 : // 片间同步 notify前卡 wait后卡
1241 38 : ret = AicpuDispatcher::SignalRecord(mainStream, preRankId, AicpuDispatcher::IPC, AicpuDispatcher::POST_SYNC);
1242 38 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("Add notify pre rank failed"), ret);
1243 38 : ret = AicpuDispatcher::SignalWait(mainStream, postRankId, AicpuDispatcher::IPC, AicpuDispatcher::POST_SYNC);
1244 38 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("Add wait post rank failed"), ret);
1245 38 : return ret;
1246 : }
1247 :
1248 60 : HcclResult TaskOrchestrator::IsSupportRDMAReduce(HcclCMDType commType, HcclDataType dataType, HcclReduceOp op)
1249 : {
1250 : static const std::set<HcclCMDType> multiThreadComTypeWhiteList = {
1251 : HcclCMDType::HCCL_CMD_BATCH_WRITE,
1252 64 : };
1253 60 : if (HcclAicpuUtils::GetBlockNum() > 1U &&
1254 60 : multiThreadComTypeWhiteList.find(commType) == multiThreadComTypeWhiteList.end()) {
1255 0 : HCCL_ERROR("Unsupported comm type %u with multi threads.", commType);
1256 0 : return HCCL_E_PARA;
1257 : }
1258 :
1259 60 : if (commType != HcclCMDType::HCCL_CMD_ALLREDUCE && commType != HcclCMDType::HCCL_CMD_REDUCE_SCATTER) {
1260 7 : return HCCL_SUCCESS;
1261 : }
1262 : static const std::set<HcclDataType> dtypeWhiteList = {
1263 : HCCL_DATA_TYPE_FP32,
1264 : HCCL_DATA_TYPE_FP16,
1265 : HCCL_DATA_TYPE_INT8,
1266 : HCCL_DATA_TYPE_INT16,
1267 : HCCL_DATA_TYPE_INT32,
1268 : HCCL_DATA_TYPE_BFP16
1269 57 : };
1270 53 : if (dtypeWhiteList.find(dataType) == dtypeWhiteList.end()) {
1271 1 : HCCL_ERROR("Unsupported datatype %s for comm type %u.", GetDataTypeEnumStr(dataType).c_str(), commType);
1272 1 : return HCCL_E_PARA;
1273 : }
1274 :
1275 : static const std::set<HcclReduceOp> reduceTypeWhiteList = {
1276 : HCCL_REDUCE_SUM,
1277 : HCCL_REDUCE_MAX,
1278 : HCCL_REDUCE_MIN
1279 54 : };
1280 52 : if (reduceTypeWhiteList.find(op) == reduceTypeWhiteList.end()) {
1281 0 : HCCL_ERROR("Unsupported reduce op %s.", GetReduceOpEnumStr(op).c_str());
1282 0 : return HCCL_E_PARA;
1283 : }
1284 52 : return HCCL_SUCCESS;
1285 : }
1286 :
1287 95 : HcclResult TaskOrchestrator::RunConcreteAlgorithm(AivAicpuOpParam *commParam, AivAicpuOpParam *commParamNext,
1288 : AicpuComContext *ctx)
1289 : {
1290 95 : void *src = reinterpret_cast<void *>(static_cast<const uintptr_t>(commParam->sendBuffer));
1291 95 : void *dst = reinterpret_cast<void *>(static_cast<const uintptr_t>(commParam->recvBuffer));
1292 95 : RECORD_PROF_TIME(hccExecStartTime);
1293 :
1294 189 : const bool waitFlag = ((ctx->devType != DevType::DEV_TYPE_310P1 && ctx->devType != DevType::DEV_TYPE_310P3) &&
1295 94 : ctx->commAlg == COMM_ALG_FULL_MESH);
1296 95 : HCCL_DEBUG("startRunAlg src:%p, dst:%p, ctx commType:%d, commParam commType:%d, waitFlag:%u.",
1297 : src, dst, ctx->commType, commParam->commType, static_cast<u32>(waitFlag));
1298 95 : if (waitFlag) {
1299 82 : CHK_RET(AicpuDispatcher::AddWaitStartTaskOnMainStream(ctx->rankId));
1300 : }
1301 :
1302 95 : HcclResult result = HCCL_SUCCESS;
1303 95 : switch (ctx->commType) {
1304 20 : case HcclCMDType::HCCL_CMD_REDUCE_SCATTER: {
1305 20 : CHK_RET(IsSupportRDMAReduce(commParam->commType, commParam->hcclDataType, commParam->opType));
1306 19 : u64 strideLen = (commParam->strideLen != 0) ? commParam->strideLen : commParam->count / ctx->rankNum;
1307 19 : AicpuReduceScatter reduceScatter(ctx);
1308 19 : result = reduceScatter.RunAlgorithm(
1309 : commParam->opType, src, dst, commParam->count, commParam->hcclDataType, strideLen);
1310 19 : break;
1311 19 : }
1312 32 : case HcclCMDType::HCCL_CMD_ALLGATHER: {
1313 32 : u64 strideLen = (commParam->strideLen != 0) ? commParam->strideLen : commParam->count;
1314 32 : AicpuAllgather allgather(ctx);
1315 32 : result = allgather.RunAlgorithm(commParam->opType, src, dst, commParam->count, commParam->hcclDataType,
1316 : strideLen, commParamNext);
1317 32 : break;
1318 32 : }
1319 33 : case HcclCMDType::HCCL_CMD_ALLREDUCE: {
1320 33 : CHK_RET(IsSupportRDMAReduce(commParam->commType, commParam->hcclDataType, commParam->opType));
1321 33 : if (ctx->determinism) {
1322 0 : u64 strideLen = (commParam->strideLen != 0) ? commParam->strideLen : commParam->count;
1323 0 : AicpuDmyCalAllreduce dmyCalAllreduce(ctx);
1324 0 : result = dmyCalAllreduce.RunAlgorithm(commParam->opType, src, dst, commParam->count,
1325 : commParam->hcclDataType, strideLen, commParamNext);
1326 0 : } else {
1327 33 : AicpuAllreduce allreduce(ctx);
1328 33 : result = allreduce.RunAlgorithm(commParam->opType, src, dst, commParam->count, commParam->hcclDataType);
1329 33 : }
1330 33 : break;
1331 : }
1332 8 : case HcclCMDType::HCCL_CMD_ALLTOALL: {
1333 8 : u64 strideLen = (commParam->strideLen != 0) ? commParam->strideLen : commParam->count;
1334 8 : AicpuAllToAll allToAll(ctx);
1335 8 : result = allToAll.RunAlgorithm(commParam->opType, src, dst, commParam->count,
1336 : commParam->hcclDataType, strideLen);
1337 8 : break;
1338 8 : }
1339 2 : default: {
1340 2 : HCCL_ERROR("commType [%d] is not supported.", commParam->commType);
1341 2 : result = HCCL_E_PARA;
1342 2 : break;
1343 : }
1344 : }
1345 :
1346 94 : ctx->curTurnCnt++;
1347 94 : HCCL_DEBUG("addEndTask, curTurnCnt:%u, totalTurnCnt:%u", ctx->curTurnCnt, ctx->totalTurnCnt);
1348 94 : if (waitFlag) {
1349 81 : CHK_RET(AicpuDispatcher::AddExecEndTaskOnMainStream(ctx->rankId));
1350 : }
1351 94 : return result;
1352 : }
|
