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