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 "aicpu_kfc_process.h"
12 :
13 : #include <numeric>
14 : #include "log_control.h"
15 : #include "common/aicpu_hccl_common.h"
16 : #include "aicpu_kfc_batchwrite_process.h"
17 : #include "aicpu_kfc_retry_process.h"
18 : #include "framework/aicpu_communicator.h"
19 : #include "algorithm/task_orchestrator.h"
20 : #include "common/aicpu_sqe_context.h"
21 : #include "dfx/mc2_trace_utils.h"
22 : #include "utils/hccl_aicpu_utils.h"
23 : #include "common/aicpu_kfc_utils.h"
24 : #include "utils/aicpu_hdc_utils.h"
25 : #include "framework/aicpu_hccl_process.h"
26 : #include "framework/aicpu_kfc_rpc_serverv2.h"
27 : #include "framework/aicpu_kfc_prof.h"
28 : #include "common/aicpu_kfc_tiling_utils.h"
29 : #include "coll_batch_write_executor.h"
30 : #include "dfx/aicpu_profiling_manager.h"
31 : #include <shared_mutex>
32 :
33 : using namespace hccl;
34 : using namespace HcclApi;
35 :
36 : ANONYMOUS_NAMESPACE_BEGIN
37 : static constexpr uint64_t KERNEL_TIMEOUT = 16 * 60;
38 : static constexpr uint64_t LOGCOUNT_PRINT_TIMEOUT = 10000;
39 : struct TimeOutCheckInfo {
40 : u64 kernelStartTime;
41 : std::unordered_map<u32, bool> msgFlag;
42 : std::unordered_map<u32, u64> msgStartTime;
43 : std::unordered_map<u32, u32> invalidMsgCount;
44 : };
45 : thread_local TimeOutCheckInfo g_timeOutInfoInst{};
46 51 : void SetMsgEnableFlag(u32 groupIdx, bool flag) {
47 51 : g_timeOutInfoInst.msgFlag[groupIdx] = flag;
48 51 : }
49 :
50 38 : bool CheckMsgEnableFlag(u32 groupIdx) {
51 38 : if (g_timeOutInfoInst.msgFlag.find(groupIdx) == g_timeOutInfoInst.msgFlag.end()) {
52 2 : return false;
53 : }
54 36 : return g_timeOutInfoInst.msgFlag[groupIdx];
55 : }
56 :
57 26 : void SetMsgStartTime(u32 groupIdx) {
58 26 : g_timeOutInfoInst.msgStartTime[groupIdx] = GetCurCpuTimestamp();
59 26 : }
60 :
61 10 : u64 GetMsgStartTime(u32 groupIdx) {
62 10 : if (g_timeOutInfoInst.msgStartTime.find(groupIdx) == g_timeOutInfoInst.msgStartTime.end()) {
63 0 : return 0UL;
64 : }
65 10 : return g_timeOutInfoInst.msgStartTime[groupIdx];
66 : }
67 :
68 14 : void SetKernelStartTime(void) {
69 14 : g_timeOutInfoInst.kernelStartTime = GetCurCpuTimestamp();
70 14 : }
71 :
72 1 : void AddMsgInValidCount(u32 groupIdx) {
73 1 : g_timeOutInfoInst.invalidMsgCount[groupIdx]++;
74 1 : }
75 :
76 24 : void ClearMsgInValidCount(u32 groupIdx) {
77 24 : g_timeOutInfoInst.invalidMsgCount[groupIdx] = 0;
78 24 : }
79 :
80 25 : uint32_t GetMsgInValidCount(u32 groupIdx) {
81 25 : if (g_timeOutInfoInst.invalidMsgCount.find(groupIdx) == g_timeOutInfoInst.invalidMsgCount.end()) {
82 8 : return 0U;
83 : }
84 17 : return g_timeOutInfoInst.invalidMsgCount[groupIdx];
85 : }
86 :
87 : struct CommInstMgr {
88 : HcclOpResParam *resParam;
89 : hccl::HcclCommAicpu *hcclCommAicpu;
90 : AicpuKfcRpcServerV2 rpcServer;
91 : };
92 :
93 : struct KfcGroupIndexInfo {
94 : std::shared_mutex mutex;
95 : u32 nextId{0U};
96 : std::unordered_map<std::string, int32_t> groupNameToId{};
97 : std::unordered_map<int32_t, CommInstMgr> instMap{};
98 : } g_commIdMap;
99 :
100 17 : int32_t InsertComIdMap(const std::string &group) {
101 17 : std::unique_lock<std::shared_mutex> rwlock(g_commIdMap.mutex);
102 17 : if (g_commIdMap.groupNameToId.find(group) == g_commIdMap.groupNameToId.end()) {
103 12 : HCCL_INFO("Insert group %s at index %u.", group.c_str(), g_commIdMap.nextId);
104 12 : g_commIdMap.groupNameToId[group] = g_commIdMap.nextId++;
105 : } else {
106 5 : HCCL_INFO("Group %s is already at index %u.", group.c_str(), g_commIdMap.groupNameToId[group]);
107 : }
108 34 : return g_commIdMap.groupNameToId[group];
109 17 : }
110 :
111 26 : int32_t GetComGroupIdx(const std::string &group) {
112 26 : std::shared_lock<std::shared_mutex> rwlock(g_commIdMap.mutex);
113 : int32_t idx;
114 26 : if (g_commIdMap.groupNameToId.find(group) == g_commIdMap.groupNameToId.end()) {
115 0 : HCCL_ERROR("Failed to find group %s in index map.", group.c_str());
116 0 : idx = -1;
117 : } else {
118 26 : idx = g_commIdMap.groupNameToId[group];
119 : }
120 26 : return idx;
121 26 : }
122 :
123 17 : HcclResult InsertCommInst(uint32_t idx, hccl::HcclCommAicpu *comm, HcclOpResParam *resParam)
124 : {
125 17 : g_commIdMap.instMap[idx].resParam = resParam;
126 17 : g_commIdMap.instMap[idx].hcclCommAicpu = comm;
127 17 : return HCCL_SUCCESS;
128 : }
129 :
130 101 : hccl::HcclCommAicpu *GetCommAicpuCommInst(uint32_t idx)
131 : {
132 101 : if (g_commIdMap.instMap.find(idx) == g_commIdMap.instMap.end()) {
133 0 : return nullptr;
134 : }
135 101 : return g_commIdMap.instMap[idx].hcclCommAicpu;
136 : }
137 :
138 25 : HcclOpResParam *GetCommAicpuResInst(uint32_t idx)
139 : {
140 25 : if (g_commIdMap.instMap.find(idx) == g_commIdMap.instMap.end()) {
141 0 : return nullptr;
142 : }
143 25 : return g_commIdMap.instMap[idx].resParam;
144 : }
145 :
146 73 : AicpuKfcRpcServerV2 *GetCommRpcServer(uint32_t idx)
147 : {
148 73 : if (g_commIdMap.instMap.find(idx) == g_commIdMap.instMap.end()) {
149 0 : return nullptr;
150 : }
151 73 : return &(g_commIdMap.instMap[idx].rpcServer);
152 : }
153 :
154 : static thread_local uint8_t g_expectPrepareId[MAX_QUE_NUM];
155 24 : void SetExpectPrepareId(uint8_t queueId, uint8_t msgId)
156 : {
157 24 : g_expectPrepareId[queueId] = msgId;
158 24 : }
159 :
160 9 : uint8_t GetExpectPrepareId(uint8_t queueId)
161 : {
162 9 : return g_expectPrepareId[queueId];
163 : }
164 :
165 : struct CommInfoCtx {
166 : AlgType algType;
167 : std::string algName;
168 : std::string tag;
169 : };
170 : static std::unordered_map<std::string, std::unordered_map<u8, CommInfoCtx>> g_commTypeInfoMap;
171 : static std::shared_mutex g_mutexForTypeInfoMap;
172 10 : void SetCommInfoCtx(const std::string &groupName, u8 commType, const CommInfoCtx &ctx)
173 : {
174 10 : std::unique_lock<std::shared_mutex> rwlock(g_mutexForTypeInfoMap);
175 10 : g_commTypeInfoMap[groupName][commType] = ctx;
176 10 : }
177 :
178 5 : HcclResult GetCommInfoCtx(const std::string &commName, u8 commType, CommInfoCtx &ctx)
179 : {
180 5 : std::shared_lock<std::shared_mutex> rwlock(g_mutexForTypeInfoMap);
181 5 : const auto groupIter = g_commTypeInfoMap.find(commName);
182 5 : if (groupIter == g_commTypeInfoMap.end()) {
183 0 : HCCL_ERROR("Failed to find group %s in type info map.", commName.c_str());
184 0 : return HCCL_E_INTERNAL;
185 : }
186 :
187 5 : const auto commIter = groupIter->second.find(commType);
188 5 : if (commIter == groupIter->second.end()) {
189 2 : HCCL_ERROR("Failed to find type %u in map for group %s.", static_cast<u32>(commType), commName.c_str());
190 2 : return HCCL_E_INTERNAL;
191 : }
192 :
193 3 : ctx = commIter->second;
194 3 : return HCCL_SUCCESS;
195 5 : }
196 :
197 : const std::unordered_map<std::string, std::string> g_algName = {
198 : {"AllGather=level0:ring", "AllGatherRingFor91093Executor"},
199 : {"AllGather=level0:fullmesh", "AllGatherMeshOpbaseExecutor"},
200 : {"AllGather=level0:doublering", "AlignedAllGatherDoubleRingFor91093Executor"},
201 : {"ReduceScatter=level0:ring", "ReduceScatterRingFor91093Executor"},
202 : {"ReduceScatter=level0:fullmesh", "ReduceScatterMeshDmaEliminationExecutor"},
203 : {"ReduceScatter=level0:doublering", "AlignedReduceScatterDoubleRingFor91093Executor"},
204 : {"AllReduce=level0:ring", "AllReduceRingFor91093Executor"},
205 : {"AllReduce=level0:fullmesh", "AllReduceMeshOpbaseLoopExecutor"},
206 : {"AllReduce=level0:doublering", "AlignedAllReduceDoubleRingFor91093Executor"},
207 : {"AlltoAll=level0:pairwise", "RunAlltoAllVStaged"},
208 : {"AlltoAll=level0:fullmesh", "RunAlltoAllDirectFullmesh"},
209 : {"BatchWrite=level0:fullmesh", BATCH_WRITE_ALG_NAME}
210 : };
211 : ANONYMOUS_NAMESPACE_END
212 :
213 : AicpuAddOneNotifyWaitSqe g_addOneNotifyWaitSqe = nullptr;
214 : AicpuAddOneRecordSqe g_addOneRecordSqe = nullptr;
215 : AicpuAddOneWriteValueRecordSqe g_addOneWriteValueRecordSqe = nullptr;
216 : AicpuAddOneMemcpySqe g_addOneMemcpySqe = nullptr;
217 : AicpuAddOneEventResetSqe g_addOneEventResetSqe = nullptr;
218 : AicpuAddOneEventRecordSqe g_addOneEventRecordSqe = nullptr;
219 : AicpuAddOneEventWaitSqe g_addOneEventWaitSqe = nullptr;
220 : AicpuAddOneRdmaDbSendSqe g_addOneRdmaDbSendSqe = nullptr;
221 : AicpuAddOneFlipPlaceHolderSqe g_addOneFlipPlaceHolderSqe = nullptr;
222 3614 : AicpuAddOneNotifyWaitSqe AicpuGetAddOneNotifyWaitSqe() { return g_addOneNotifyWaitSqe; }
223 2079 : AicpuAddOneRecordSqe AicpuGetAddOneRecordSqe() { return g_addOneRecordSqe; }
224 1584 : AicpuAddOneWriteValueRecordSqe AicpuGetAddOneWriteValueRecordSqe() { return g_addOneWriteValueRecordSqe; }
225 854 : AicpuAddOneMemcpySqe AicpuGetAddOneMemcpySqe() { return g_addOneMemcpySqe; }
226 2 : AicpuAddOneEventResetSqe AicpuGetAddOneEventResetSqe() { return g_addOneEventResetSqe; }
227 0 : AicpuAddOneEventRecordSqe AicpuGetAddOneEventRecordSqe() { return g_addOneEventRecordSqe; }
228 3 : AicpuAddOneEventWaitSqe AicpuGetAddOneEventWaitSqe() { return g_addOneEventWaitSqe; }
229 1 : AicpuAddOneRdmaDbSendSqe AicpuGetAddOneRdmaDbSendSqe() { return g_addOneRdmaDbSendSqe; }
230 1 : AicpuAddOneFlipPlaceHolderSqe AicpuGetAddOneFlipPlaceHolderSqe() { return g_addOneFlipPlaceHolderSqe; }
231 :
232 : ANONYMOUS_NAMESPACE_BEGIN
233 80 : void InitSqCqFun(AicpuComContext *ctx)
234 : {
235 80 : if (ctx->devType == DevType::DEV_TYPE_310P1 || ctx->devType == DevType::DEV_TYPE_310P3) {
236 0 : g_addOneNotifyWaitSqe = AddOneNotifyWaitSqeV2;
237 0 : g_addOneRecordSqe = AddOneRecordSqeV2;
238 0 : g_addOneWriteValueRecordSqe = AddOneWriteValueRecordSqeV2;
239 0 : g_addOneMemcpySqe = AddOneMemcpySqeV2;
240 0 : g_addOneEventResetSqe = AddOneEventResetSqeV2;
241 0 : g_addOneEventRecordSqe = AddOneEventRecordSqeV2;
242 0 : g_addOneEventWaitSqe = AddOneEventWaitSqeV2;
243 : } else {
244 80 : g_addOneNotifyWaitSqe = AddOneNotifyWaitSqeV1;
245 80 : g_addOneRecordSqe = AddOneRecordSqeV1;
246 80 : g_addOneWriteValueRecordSqe = AddOneWriteValueRecordSqeV1;
247 80 : g_addOneMemcpySqe = AddOneMemcpySqeV1;
248 80 : g_addOneEventResetSqe = AddOneEventResetSqeV1;
249 80 : g_addOneEventRecordSqe = AddOneEventRecordSqeV1;
250 80 : g_addOneEventWaitSqe = AddOneEventWaitSqeV1;
251 80 : g_addOneFlipPlaceHolderSqe = AddOneFlipPlaceHolderSqeV1;
252 80 : g_addOneRdmaDbSendSqe = AddOneRdmaDbSendSqeV1;
253 : }
254 80 : }
255 :
256 0 : HcclResult InitIbversData(HccCommResParamTask *commParam, AicpuComContext *ctx) {
257 0 : HCCL_INFO("commParam->ibverbsData:%llu", commParam->ibverbsData);
258 0 : if (commParam->ibverbsDataSize != static_cast<u64>(ctx->rankNum) * sizeof(TransportDeviceNormalData)) {
259 0 : HCCL_ERROR("ibverbsData size[%llu] is not valid, expect size[%llu]",
260 : commParam->ibverbsDataSize, static_cast<u64>(ctx->rankNum) * sizeof(TransportDeviceNormalData));
261 0 : return HCCL_E_PARA;
262 : }
263 0 : ctx->ibversData.resize(ctx->rankNum);
264 0 : for (u32 i = 0; i < ctx->rankNum; i++) {
265 0 : void *memPtr = reinterpret_cast<void *>(commParam->ibverbsData + sizeof(TransportDeviceNormalData) * i);
266 0 : ctx->ibversData[i] = *(static_cast<TransportDeviceNormalData *>(memPtr));
267 0 : ctx->ibversData[i].Print();
268 : }
269 0 : return HCCL_SUCCESS;
270 : }
271 :
272 80 : void InitRankInfo(HccCommResParamTask *commParam, AicpuComContext *ctx)
273 : {
274 720 : for (u32 i = 0; i < ctx->rankNum; i++) {
275 640 : ctx->rankInfo[i].rankId = i;
276 640 : ctx->rankInfo[i].window = commParam->windowsIn[i];
277 640 : ctx->rankInfo[i].windowOut = commParam->windowsOut[i];
278 : }
279 80 : }
280 :
281 : template <typename T>
282 3360 : HcclResult InitAndVerifySignal(const HcclSignalInfo &signalInfo, std::shared_ptr<T> ¬ify, u64 &addr)
283 : {
284 3360 : if (signalInfo.resId == INVALID_U64) {
285 0 : HCCL_INFO("[HcclCommAicpu][%s] resId is invalid, need not check", __func__);
286 0 : return HCCL_SUCCESS;
287 : }
288 :
289 3360 : EXCEPTION_CATCH((notify = std::make_shared<T>()), return HCCL_E_PTR);
290 3360 : CHK_SMART_PTR_NULL(notify);
291 3360 : CHK_RET(notify->Init(signalInfo, NotifyLoadType::DEVICE_NOTIFY));
292 : HcclSignalInfo notifyInfo;
293 3360 : CHK_RET(notify->GetNotifyData(notifyInfo));
294 3360 : addr = notifyInfo.addr;
295 3360 : HCCL_INFO("[HcclCommAicpu][%s] success, resId[%u], tsId:%d, devId[%u]", __func__, signalInfo.resId,
296 : signalInfo.tsId, signalInfo.devId);
297 3360 : return HCCL_SUCCESS;
298 : }
299 :
300 80 : HcclResult InitSignalInfo(HccCommResParamTask *commParam, AicpuComContext *ctx)
301 : {
302 720 : for (u32 i = 0; i < ctx->rankNum; i++) {
303 : // 跨片notify只用在其它rank上，本片位置未填写有效值
304 640 : if (ctx->rankId == i) {
305 80 : continue;
306 : }
307 :
308 : // no ipc pre sync
309 560 : u64 address = 0;
310 560 : std::shared_ptr<LocalNotify> localNotify;
311 560 : HcclSignalInfo *sigInfo = &commParam->signalInfo.noIpcNotifys[i];
312 560 : CHK_RET(InitAndVerifySignal(*sigInfo, localNotify, address));
313 560 : ctx->noIpcPreNotify[i].actualNotifyId = static_cast<s32>(sigInfo->resId);
314 :
315 560 : if (sigInfo->rankId != ctx->rankInfo[i].rankId) {
316 0 : HCCL_DEBUG("rankId mismatch. current process rank:%d, sigInfo rank:%d", ctx->rankInfo[i].rankId,
317 : sigInfo->rankId);
318 0 : return HCCL_E_INTERNAL;
319 : }
320 :
321 : // no ipc post sync
322 560 : sigInfo = &commParam->signalInfo.noIpcNotifys[ctx->rankNum + i];
323 560 : CHK_RET(InitAndVerifySignal(*sigInfo, localNotify, address));
324 560 : ctx->noIpcPostNotify[i].actualNotifyId = static_cast<s32>(sigInfo->resId);
325 :
326 : // ipc pre record
327 560 : sigInfo = &commParam->signalInfo.ipcNotifys[i];
328 560 : std::shared_ptr<RemoteNotify> remoteNotify;
329 560 : CHK_RET(InitAndVerifySignal(*sigInfo, remoteNotify, ctx->ipcPreRecordNotify[i].address));
330 560 : ctx->ipcPreRecordNotify[i].actualNotifyId = static_cast<s32>(sigInfo->resId);
331 :
332 : // ipc pre wait
333 560 : sigInfo = &commParam->signalInfo.ipcNotifys[ctx->rankNum + i];
334 560 : CHK_RET(InitAndVerifySignal(*sigInfo, localNotify, ctx->ipcPreWaitNotify[i].address));
335 560 : ctx->ipcPreWaitNotify[i].actualNotifyId = static_cast<s32>(sigInfo->resId);
336 :
337 : // ipc post record
338 560 : sigInfo = &commParam->signalInfo.ipcNotifys[2 * ctx->rankNum + i]; // 2 is ipc post record(8-15)
339 560 : CHK_RET(InitAndVerifySignal(*sigInfo, remoteNotify, ctx->ipcPostRecordNotify[i].address));
340 560 : ctx->ipcPostRecordNotify[i].actualNotifyId = static_cast<s32>(sigInfo->resId);
341 :
342 : // ipc post wait
343 560 : sigInfo = &commParam->signalInfo.ipcNotifys[3 * ctx->rankNum + i]; // 3 is ipc post wait(16-23)
344 560 : CHK_RET(InitAndVerifySignal(*sigInfo, localNotify, ctx->ipcPostWaitNotify[i].address));
345 560 : ctx->ipcPostWaitNotify[i].actualNotifyId = static_cast<s32>(sigInfo->resId);
346 560 : }
347 80 : return HCCL_SUCCESS;
348 : }
349 :
350 80 : HcclResult InitEventId(HccCommResParamTask *commParam, AicpuComContext *ctx)
351 : {
352 720 : for (u32 i = 0; i < ctx->rankNum; i++) {
353 : // eventid只用在片内，放全局
354 640 : HcclSignalInfo *sigInfo = &commParam->signalInfo.noIpcEvents[i];
355 640 : if (sigInfo->rankId == ctx->rankId) {
356 : // 盘古230B入图场景连续跑第二次会出现eventId校验失败，当前不使用event，删除KfcResIsInvalid校验
357 80 : ctx->eventIds[i] = sigInfo->resId;
358 : }
359 : }
360 80 : return HCCL_SUCCESS;
361 : }
362 :
363 80 : HcclResult InitAicpuOpNotify(HccCommResParamTask *commParam, AicpuComContext *ctx)
364 : {
365 240 : for (u32 i = 0; i < sizeof(ctx->aicpuOpNotify) / sizeof(ctx->aicpuOpNotify[0]); i++) {
366 160 : HcclSignalInfo *sigInfo = &commParam->signalInfo.aicpuOpNotify[i];
367 160 : std::shared_ptr<LocalNotify> localNitfy;
368 160 : EXCEPTION_CATCH((localNitfy = std::make_shared<LocalNotify>()), return HCCL_E_PTR);
369 160 : CHK_RET(localNitfy->Init(*sigInfo, NotifyLoadType::DEVICE_NOTIFY));
370 : HcclSignalInfo signalInfo;
371 160 : CHK_RET(localNitfy->GetNotifyData(signalInfo));
372 160 : ctx->aicpuOpNotify[i].actualNotifyId = static_cast<s32>(sigInfo->resId);
373 160 : ctx->aicpuOpNotify[i].address = signalInfo.addr;
374 160 : }
375 80 : return HCCL_SUCCESS;
376 : }
377 :
378 80 : HcclResult InitTimeOutConfig(HccCommResParamTask *commParam, AicpuComContext *ctx)
379 : {
380 80 : ctx->dfxExtendInfo.dfxTimeOutConfig.sqeTimeOutTimeOut = commParam->config.notifyWaitTime;
381 80 : ctx->dfxExtendInfo.dfxTimeOutConfig.sqeCreditTimeOut = RT_STARS_NEVER_TIMEOUT_KERNEL_CREDIT;
382 80 : ctx->dfxExtendInfo.dfxTimeOutConfig.sqeWaitTimeOut = dfx::kKfcTimeOut;
383 80 : ctx->dfxExtendInfo.dfxTimeOutConfig.sqFullWaitTimeOut = dfx::kSqFullWaitTimeOut;
384 80 : HCCL_INFO("DFX timeout config init successfully with details: [%s]",
385 : ctx->dfxExtendInfo.dfxTimeOutConfig.ToString().c_str());
386 80 : return HCCL_SUCCESS;
387 : }
388 :
389 80 : HcclResult InitChipType(AicpuComContext *ctx)
390 : {
391 80 : CHK_RET(hrtHalGetDeviceType(ctx->devId, ctx->devType));
392 80 : CHK_RET(hrtHalGetDeviceInfo(ctx->devId, MODULE_TYPE_SYSTEM, INFO_TYPE_PHY_CHIP_ID, &ctx->chipId));
393 80 : if (ctx->devType == DevType::DEV_TYPE_910 || ctx->devType == DevType::DEV_TYPE_NOSOC ||
394 80 : ctx->devType == DevType::DEV_TYPE_COUNT) {
395 0 : HCCL_ERROR("Get devtype [%d] is invalid", ctx->devType);
396 0 : return HCCL_E_DRV;
397 : }
398 80 : if (ctx->devType == DevType::DEV_TYPE_310P3 || ctx->devType == DevType::DEV_TYPE_310P1) {
399 : uint32_t ssid;
400 0 : const HcclResult ret = hrtDrvMemSmmuQuery(ctx->devId, &ssid);
401 0 : HCCL_DEBUG("ssid %u", ssid);
402 0 : ctx->ssid = ssid;
403 0 : ctx->determinism = false;
404 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("hrtDrvMemSmmuQuery error"), HCCL_E_DRV);
405 : }
406 80 : InitSqCqFun(ctx);
407 80 : return HCCL_SUCCESS;
408 : }
409 :
410 1 : void GetNextMsgFromMsg(AivAicpuOpParam *msg, AivAicpuOpParam *nextMsg, u64 dataLen, u32 rankNum)
411 : {
412 1 : *(nextMsg) = *(msg);
413 : // nextMsg的偏移同UpdateMsg
414 1 : if (nextMsg->commType == HcclCMDType::HCCL_CMD_REDUCE_SCATTER) {
415 0 : nextMsg->sendBuffer = nextMsg->sendBuffer + dataLen / rankNum;
416 0 : nextMsg->recvBuffer = nextMsg->recvBuffer + dataLen / rankNum;
417 : } else {
418 1 : nextMsg->sendBuffer = nextMsg->sendBuffer + dataLen;
419 1 : nextMsg->recvBuffer = nextMsg->recvBuffer + dataLen;
420 : }
421 2 : nextMsg->PrintMsg("nextMsg");
422 1 : }
423 :
424 75 : void GetCommonHcclMsg(HcclMsg *hcclMsg, CommonHcclMsg *commonHcclMsg, u64 tilingBase)
425 : {
426 75 : const HcclTilingVersion ver = hcclMsg->addMsg.v0Msg.version;
427 75 : if (ver != HcclTilingVersion::DEPRECATED_TILING_VERSION) {
428 13 : const size_t copyOffset = offsetof(HcclMsg, addMsg);
429 13 : (void)memcpy_s(commonHcclMsg, copyOffset, hcclMsg, copyOffset);
430 13 : if (ver == HcclTilingVersion::ONLINE_COMPILATION_TILING_VERSION) {
431 0 : commonHcclMsg->ccOpTilingData = hcclMsg->addMsg.v1Msg.ccOpTilingData + tilingBase;
432 : } else {
433 13 : commonHcclMsg->ccOpTilingData = hcclMsg->addMsg.v1Msg.ccOpTilingData;
434 : }
435 13 : commonHcclMsg->valid = hcclMsg->addMsg.v1Msg.valid;
436 13 : commonHcclMsg->hcclDataType = static_cast<HcclDataType>(hcclMsg->addMsg.v1Msg.hcclDataType);
437 13 : commonHcclMsg->repeatCnt = hcclMsg->addMsg.v1Msg.repeatCnt;
438 13 : commonHcclMsg->selfHandleID = hcclMsg->addMsg.v1Msg.selfHandleID;
439 13 : commonHcclMsg->seqNum = hcclMsg->addMsg.v1Msg.seqNum;
440 13 : commonHcclMsg->version = hcclMsg->addMsg.v1Msg.version;
441 13 : commonHcclMsg->xorCheck = hcclMsg->addMsg.v1Msg.xorCheck;
442 : } else {
443 62 : (void)memcpy_s(commonHcclMsg, sizeof(HcclMsg), hcclMsg, sizeof(HcclMsg));
444 62 : commonHcclMsg->ccOpTilingData = 0UL;
445 : }
446 75 : }
447 :
448 95 : AicpuCCExecOp GetCcOpType(u64 comDataLen, u64 rankNum)
449 : {
450 : AicpuCCExecOp ccType;
451 95 : AicpuComContext *ctx = AicpuGetComContext();
452 95 : if (ctx->devType == DevType::DEV_TYPE_310P1 || ctx->devType == DevType::DEV_TYPE_310P3) {
453 1 : if (ctx->onlyRead > 0) {
454 0 : HCCL_DEBUG("Only read mode enabled");
455 0 : ccType = CC_EXE_ONE_SHOT_SINGLE_RING;
456 1 : } else if (rankNum == 2) { // 2 卡
457 0 : if (comDataLen < HCCL_SMALL_COUNT_1_M) {
458 0 : ccType = CC_EXE_ONE_SHOT_1_STREAM;
459 : } else {
460 0 : ccType = CC_EXE_TWO_SHOT_1_STREAM;
461 : }
462 : } else { // 2 卡以上
463 1 : if (comDataLen < HCCL_SMALL_COUNT_256K && (rankNum & (rankNum - 1)) == 0) {
464 1 : ccType = CC_EXE_ONE_SHOT_HD;
465 : } else {
466 0 : ccType = CC_EXE_ONE_SHOT_SINGLE_RING;
467 : }
468 : }
469 : } else {
470 94 : if ((comDataLen < AC_DEFAULT_ONE_SHOT_SIZE) && ((rankNum % AC_DEFAULT_RANK_GROUP) == 0)) {
471 79 : ccType = CC_EXE_ONE_SHOT_8_STREAM;
472 : } else {
473 15 : ccType = CC_EXE_TWO_SHOT_8_STREAM;
474 : }
475 : }
476 95 : return ccType;
477 : }
478 :
479 23 : void UpdateMsg(AivAicpuOpParam *msg, u64 dataLen, u32 rankNum)
480 : {
481 : // 如果是reduceScatter算法，sendBuffer和recvBuffer的偏移为recvCnt，即sendCnt/rankNum
482 : // allgather和allreduce算法，sendBuffer和recvBuffer的偏移为recvCnt=sendCnt
483 : // all2all算法，sendBuffer和recvBuffer的偏移为 sendCnt / rankNum
484 : // 如果recvBuffer是非连续存储的，则recvBuffer的偏移将变更为 sendCnt
485 23 : if (msg->commType == HcclCMDType::HCCL_CMD_REDUCE_SCATTER) {
486 1 : msg->sendBuffer = msg->sendBuffer + dataLen / rankNum;
487 1 : msg->recvBuffer = msg->recvBuffer + dataLen / rankNum;
488 : } else {
489 22 : msg->sendBuffer = msg->sendBuffer + dataLen;
490 22 : msg->recvBuffer = msg->recvBuffer + dataLen;
491 : }
492 23 : if (msg->commType == HcclCMDType::HCCL_CMD_ALLREDUCE || msg->commType == HcclCMDType::HCCL_CMD_ALLTOALL) {
493 14 : msg->winOffset = msg->winOffset + dataLen;
494 : }
495 46 : msg->PrintMsg("update msg");
496 23 : }
497 :
498 20 : HcclResult SetMsgWinOffset(AicpuComContext *ctx, AivAicpuOpParam *msg)
499 : {
500 20 : if (msg->useBufferType == MC2_BUFFER_TYPE_WINDOW_IN &&
501 2 : ((msg->commType == HcclCMDType::HCCL_CMD_ALLREDUCE && !ctx->determinism) ||
502 2 : msg->commType == HcclCMDType::HCCL_CMD_ALLTOALL)) {
503 : // sendBuffer 减去本卡的winIn
504 2 : AicpuComRankInfo *selfRankInfo = &ctx->rankInfo[ctx->rankId];
505 2 : if (msg->sendBuffer < selfRankInfo->window) {
506 0 : HCCL_ERROR("sendBuffer addr[%p] must bigger than window addr[%p].", msg->sendBuffer,
507 : selfRankInfo->window);
508 0 : return HCCL_E_PARA;
509 : }
510 2 : msg->winOffset = msg->sendBuffer - selfRankInfo->window;
511 : }
512 20 : HCCL_INFO("Offsetting winOffset %lu", msg->winOffset);
513 20 : return HCCL_SUCCESS;
514 : }
515 :
516 36 : bool CheckNsCommand(hccl::HcclCommAicpu *comm) {
517 : KfcCommand cmd;
518 36 : if (comm->BackGroundGetCmd(cmd) != HCCL_SUCCESS || cmd != KfcCommand::NsStopLaunch) {
519 34 : return false;
520 : }
521 2 : comm->SetNsStopLaunchStatus(true);
522 2 : HCCL_WARNING("N second stop Launch for recv stop launch cmd.");
523 2 : return true;
524 : }
525 :
526 21 : HcclResult CheckNsStopLaunchStatus(const std::vector<u32> &groupIds)
527 : {
528 48 : for (const auto i: groupIds) {
529 28 : hccl::HcclCommAicpu *comm = GetCommAicpuCommInst(i);
530 28 : if (comm != nullptr && comm->GetNsStopLaunchStatus()) {
531 1 : return HCCL_E_SUSPENDING;
532 : }
533 : }
534 20 : return HCCL_SUCCESS;
535 : }
536 :
537 12 : bool GetOpRetryEnable(const std::vector<u32> &groupIds)
538 : {
539 13 : for (const auto i: groupIds) {
540 12 : hccl::HcclCommAicpu *comm = GetCommAicpuCommInst(i);
541 12 : if (comm == nullptr || !comm->GetOpRetryEnable()) {
542 11 : return false;
543 : }
544 : }
545 1 : return true;
546 : }
547 :
548 38 : HcclResult CheckRestartError(hccl::HcclCommAicpu *comm) {
549 : // 支持重执行时，检测是否有可重执行的sdma异常, 或者kStopLaunch命令
550 38 : if (comm->GetOpRetryEnable()) {
551 3 : if (comm->IsTaskExceptionForHccs()) {
552 1 : HCCL_WARNING("MC2 restart Sdma error happened.");
553 2 : return HCCL_E_SUSPENDING;
554 : }
555 :
556 2 : KfcCommand cmd = KfcCommand::kNone;
557 2 : CHK_RET(comm->BackGroundGetCmd(cmd));
558 2 : if (cmd == KfcCommand::kStopLaunch) {
559 1 : HCCL_WARNING("MC2 restart receive kfc command stop launch.");
560 1 : return HCCL_E_SUSPENDING;
561 : }
562 : }
563 36 : return HCCL_SUCCESS;
564 : }
565 :
566 : static constexpr u32 LOG_INTERVAL = 10000U;
567 11 : HcclResult CheckFinishByStream(HcclCommAicpu &comm, size_t streamIdx, bool tailQueryFlag = true)
568 : {
569 : uint32_t sqHead, sqTail;
570 11 : Stream &stream = (streamIdx == SIZE_MAX ? comm.GetMainStream() : comm.GetSlaveStream()[streamIdx]);
571 11 : const uint32_t sqId = stream.sqId();
572 11 : if (tailQueryFlag) {
573 9 : CHK_RET(QuerySqStatusByType(comm.GetDevId(), sqId, DRV_SQCQ_PROP_SQ_TAIL, sqTail));
574 : } else {
575 2 : sqTail = stream.GetSqeContextPtr()->buffer.sqTail;
576 : }
577 11 : CHK_RET(QuerySqStatusByType(comm.GetDevId(), sqId, DRV_SQCQ_PROP_SQ_HEAD, sqHead));
578 11 : if (sqTail == sqHead) {
579 11 : HCCL_DEBUG("Stream %u finished, sq id %u, head&tail %u.", stream.id(), stream.sqId(), sqHead);
580 11 : return HCCL_SUCCESS;
581 : }
582 :
583 : static uint32_t logHead = UINT32_MAX;
584 : static uint32_t logTail = UINT32_MAX;
585 : static uint32_t loopCnt;
586 0 : if (++loopCnt % LOG_INTERVAL == 0U) {
587 0 : if (logHead != sqHead || logTail != sqTail) {
588 0 : logHead = sqHead;
589 0 : logTail = sqTail;
590 0 : HCCL_RUN_INFO("Current state. devId:%u sqid:%d, head:%u, tail:%u, group[%s]",
591 : comm.GetDevId(), sqId, sqHead, sqTail, comm.GetGroupName().c_str());
592 : }
593 : }
594 0 : return HCCL_E_UNAVAIL;
595 : }
596 :
597 37 : HcclResult RpcServerPreCheck(AicpuKfcRpcServerV2 *rpc, hccl::HcclCommAicpu *comm, bool &finalizeFlag)
598 : {
599 37 : if (CheckNsCommand(comm)) {
600 2 : return HCCL_E_SUSPENDING;
601 : }
602 35 : if (CheckRestartError(comm) == HCCL_E_SUSPENDING) {
603 0 : return HCCL_E_SUSPENDING;
604 : }
605 35 : if (comm->GetDfxExtendInfo()->pollStatus == PollStatus::kStopAsException) {
606 2 : if (comm->GetOpRetryEnable() && comm->IsTaskExceptionForHccs()) {
607 1 : HCCL_WARNING("MC2 restart Sdma error happened.");
608 1 : return HCCL_E_SUSPENDING;
609 : }
610 1 : HCCL_ERROR("MC2 hccl aicpu exec failed, for task exception.");
611 1 : return HCCL_E_INTERNAL;
612 : }
613 33 : if (rpc->GetIsFinalize()) {
614 9 : if (CheckFinishByStream(*comm, SIZE_MAX) == HCCL_SUCCESS) {
615 9 : finalizeFlag = true;
616 9 : rpc->WriteFinishWhenAllFinalize();
617 : }
618 9 : return HCCL_E_AGAIN;
619 : }
620 24 : return HCCL_SUCCESS;
621 : }
622 :
623 : static constexpr u64 BARRIER_TIMEOUT = static_cast<u64>(NSEC_PER_SEC) * 60UL;
624 26 : HcclResult BarrierProcess(u32 groupIdx, u32 localGroupIdx, u32 queueId, BarrierStatus &status)
625 : {
626 26 : AicpuKfcRpcServerV2 *rpc = GetCommRpcServer(groupIdx);
627 26 : BarrierInfo *barrierInfos = rpc->GetBarrierInfoByGroupIdx(localGroupIdx);
628 26 : BarrierStatus &selfFlag = barrierInfos[queueId].status;
629 26 : if (selfFlag == BarrierStatus::NO_BARRIER) {
630 24 : barrierInfos[queueId].lastTimeStamp = GetCurCpuTimestamp();
631 24 : status = BarrierStatus::NO_BARRIER;
632 24 : return HCCL_SUCCESS;
633 : }
634 :
635 2 : u32 &barrierFinishCnt = rpc->GetBarrierFinishCnts()[HcclAicpuUtils::GetBlockIdx()];
636 2 : if (selfFlag == BarrierStatus::SELF_BARRIER) {
637 2 : if (CheckFinishByStream(*GetCommAicpuCommInst(groupIdx), queueId, false) == HCCL_SUCCESS) {
638 2 : barrierInfos[queueId].lastTimeStamp = GetCurCpuTimestamp();
639 2 : selfFlag = BarrierStatus::INTER_BARRIER;
640 2 : ++barrierFinishCnt;
641 2 : HCCL_INFO("[%s][Queue %u]All tasks in queue are finished in block %u, finish count %u.",
642 : __func__, queueId, HcclAicpuUtils::GetBlockIdx(), barrierFinishCnt);
643 : }
644 : }
645 :
646 2 : if (selfFlag == BarrierStatus::INTER_BARRIER) {
647 2 : u32 start = 0U;
648 2 : u32 end = 0U;
649 2 : rpc->GetLocalQueueRange(start, end);
650 2 : if (barrierFinishCnt == end + 1U - start) {
651 2 : CHK_PRT_RET(AicpuKfcUtils::ThreadBarrier(BARRIER_TIMEOUT) != HCCL_SUCCESS,
652 : HCCL_ERROR("[%s]Failed to wait in block %u, finish count %u.",
653 : __func__, HcclAicpuUtils::GetBlockIdx(), barrierFinishCnt),
654 : HCCL_E_AGAIN);
655 1 : rpc->ClearBarrierStatus(localGroupIdx, start, barrierFinishCnt);
656 1 : barrierFinishCnt = 0U;
657 1 : return HCCL_SUCCESS;
658 : }
659 : }
660 :
661 1 : status = selfFlag;
662 1 : const u64 ts = GetCurCpuTimestamp();
663 1 : CHK_PRT_RET(ts - barrierInfos[queueId].lastTimeStamp > BARRIER_TIMEOUT,
664 : HCCL_ERROR("[%s]Timeout when checking queue %u, finish count %u.",
665 : __func__, queueId, barrierFinishCnt),
666 : HCCL_E_AGAIN);
667 :
668 1 : return HCCL_SUCCESS;
669 : }
670 :
671 10 : void FinalizeProcess(u32 queueIdx, hccl::HcclCommAicpu &commAicpu, AicpuKfcRpcServerV2 &rpcServer)
672 : {
673 10 : if (AicpuKfcProf::IsDebugModeEquals(MC2_DEBUG_PRINT_BUFF)) {
674 10 : rpcServer.PrintAllHcclMsgAreaData();
675 : }
676 10 : rpcServer.SetIsFinalize(queueIdx, true);
677 10 : if (rpcServer.GetTotalQueueNum() == 0U) {
678 8 : rpcServer.ResetCommitTaskAdd(commAicpu.GetDispatcher(), &(commAicpu.GetMainStream()));
679 8 : LaunchTask(commAicpu.GetDispatcher(), commAicpu.GetMainStream());
680 : }
681 10 : SetExpectPrepareId(queueIdx, 0U);
682 10 : }
683 :
684 3 : HcclResult AddTaskForGroupSyncMsg(const std::vector<u32> &groupIds, u32 localGroupIdx, CommonHcclMsg *hcclMsg)
685 : {
686 3 : if (static_cast<uint32_t>(hcclMsg->commDepGroupID) == localGroupIdx) {
687 0 : HCCL_ERROR("InterHcclGroupSync must be used for cross-domain synchronization, group id %d",
688 : hcclMsg->commDepGroupID);
689 0 : return HCCL_E_INTERNAL;
690 : }
691 :
692 3 : CHK_PRT_RET(static_cast<size_t>(hcclMsg->commDepGroupID) >= groupIds.size(),
693 : HCCL_ERROR("Invalid group id %d.", hcclMsg->commDepGroupID), HCCL_E_INTERNAL);
694 :
695 3 : AicpuKfcRpcServerV2 *rpcServerDep = GetCommRpcServer(groupIds[hcclMsg->commDepGroupID]);
696 3 : if (rpcServerDep == nullptr) {
697 0 : HCCL_ERROR("get rpc server failed, group id %d", hcclMsg->commDepGroupID);
698 0 : return HCCL_E_INTERNAL;
699 : }
700 3 : uint64_t waitAddr = rpcServerDep->GetFinishAddrByHandleId(hcclMsg->commDepHandleID);
701 3 : if (waitAddr == 0) {
702 2 : HCCL_INFO("%s waitAddr is not ready, group id %d", __func__, hcclMsg->commDepGroupID);
703 2 : return HCCL_E_UNAVAIL;
704 : }
705 1 : int32_t turnNum = rpcServerDep->GetMsgRepeatCnt(hcclMsg->commDepHandleID);
706 1 : if (turnNum < 0) {
707 0 : HCCL_INFO("%s comm group %d idx %d is not ready", __func__, hcclMsg->commDepGroupID, hcclMsg->commDepHandleID);
708 0 : return HCCL_E_UNAVAIL;
709 : }
710 :
711 1 : const u32 groupIdx = groupIds[localGroupIdx];
712 1 : hccl::HcclCommAicpu *commAicpu = GetCommAicpuCommInst(groupIdx);
713 1 : AicpuKfcRpcServerV2 *rpcServer = GetCommRpcServer(groupIdx);
714 1 : CHK_PRT_RET(commAicpu == nullptr || rpcServer == nullptr,
715 : HCCL_ERROR("Invalid group index %u.", groupIdx), HCCL_E_INTERNAL);
716 1 : rpcServer->SetNeedRetryFlag(false);
717 1 : CHK_RET(rpcServer->AddCcoreWait(commAicpu->GetDispatcher(), waitAddr, static_cast<uint32_t>(turnNum),
718 : &(commAicpu->GetMainStream()), false));
719 1 : return HCCL_SUCCESS;
720 : }
721 :
722 8 : void PrepareOpParam(hccl::OpParam *opParam, CommonHcclMsg *hcclMsg, AicpuKfcRpcServerV2 &rpc,
723 : hccl::HcclCommAicpu *commAicpu)
724 : {
725 8 : if (AicpuKfcProf::IsDebugModeEquals(MC2_DEBUG_SDMA_ERROR)) {
726 0 : opParam->inputPtr = reinterpret_cast<void *>(0xdeadbeef);
727 0 : opParam->outputPtr = reinterpret_cast<void *>(0xdeadbeef);
728 : } else {
729 8 : opParam->inputPtr = reinterpret_cast<void *>(hcclMsg->sendBuffer);
730 8 : opParam->outputPtr = reinterpret_cast<void *>(hcclMsg->recvBuffer);
731 : }
732 8 : opParam->reduceType = hcclMsg->opType;
733 8 : opParam->stream = commAicpu->GetMainStream();
734 8 : opParam->syncMode = SyncMode::DEFAULT_TIMEWAITSYNCMODE;
735 8 : opParam->opBaseAtraceInfo = nullptr;
736 8 : opParam->opType = static_cast<HcclCMDType>(hcclMsg->commType);
737 8 : if (hcclMsg->commType == HcclCMDType::HCCL_CMD_ALLTOALLV || hcclMsg->commType == HcclCMDType::HCCL_CMD_ALLTOALL) {
738 4 : HcclMsgExt *hcclMsgExt = rpc.GetHcclMsgExtPtr();
739 4 : opParam->All2AllDataDes.sendType = opParam->All2AllDataDes.recvType = hcclMsg->hcclDataType;
740 4 : opParam->All2AllDataDes.sendCount = hcclMsg->dataCnt;
741 4 : if (hcclMsg->commType == HcclCMDType::HCCL_CMD_ALLTOALL && hcclMsg->strideCount > 0UL) {
742 9 : for (uint32_t i = 0U; i < commAicpu->GetRankSize(); ++i) {
743 8 : hcclMsgExt->sendCounts[i] = hcclMsgExt->recvCounts[i] = hcclMsg->dataCnt;
744 8 : hcclMsgExt->sendOffset[i] = hcclMsgExt->recvOffset[i] = hcclMsg->strideCount * i;
745 : }
746 1 : opParam->opType = static_cast<HcclCMDType>(HcclCMDType::HCCL_CMD_ALLTOALLV);
747 : }
748 4 : if (opParam->opType == static_cast<HcclCMDType>(HcclCMDType::HCCL_CMD_ALLTOALLV)) {
749 3 : opParam->All2AllDataDes.sendCounts = static_cast<void *>(hcclMsgExt->sendCounts);
750 3 : opParam->All2AllDataDes.recvCounts = static_cast<void *>(hcclMsgExt->recvCounts);
751 3 : opParam->All2AllDataDes.sdispls = static_cast<void *>(hcclMsgExt->sendOffset);
752 3 : opParam->All2AllDataDes.rdispls = static_cast<void *>(hcclMsgExt->recvOffset);
753 : }
754 8 : } else if (hcclMsg->commType == HcclCMDType::HCCL_CMD_BATCH_WRITE) {
755 3 : opParam->BatchWriteDataDes.itemNum = hcclMsg->dataCnt;
756 3 : opParam->BatchWriteDataDes.queueNum = rpc.GetTotalQueueNum();
757 3 : opParam->BatchWriteDataDes.queueIdx = static_cast<u32>(hcclMsg->opType);
758 3 : HCCL_DEBUG("[Sdma-BatchWrite]Queue size %u, global queue id %u, item number %u.",
759 : opParam->BatchWriteDataDes.queueNum, opParam->BatchWriteDataDes.queueIdx,
760 : opParam->BatchWriteDataDes.itemNum);
761 : } else {
762 1 : const u64 totalSize = hcclMsg->dataCnt * DataUnitSize(hcclMsg->hcclDataType);
763 1 : opParam->DataDes.count = hcclMsg->dataCnt;
764 1 : opParam->DataDes.dataType = hcclMsg->hcclDataType;
765 1 : opParam->DataDes.strideCount = hcclMsg->strideCount;
766 1 : opParam->inputSize = totalSize;
767 1 : opParam->outputSize = totalSize;
768 : }
769 8 : }
770 :
771 10 : bool SelectAlgName(const std::string &algConfig, u32 topoType, std::string &algName)
772 : {
773 10 : std::string curConfig;
774 10 : std::size_t found = algConfig.find(";");
775 10 : if (found == 0) {
776 0 : return false;
777 10 : } else if (found == std::string::npos) {
778 5 : curConfig = algConfig;
779 : } else {
780 5 : curConfig = algConfig.substr(0, found);
781 : }
782 10 : if (static_cast<TopoType>(topoType) == TopoType::TOPO_TYPE_NP_SINGLE_RING) {
783 1 : if (curConfig == "AllGather=level0:doublering" || curConfig == "ReduceScatter=level0:doublering" ||
784 0 : curConfig == "AllReduce=level0:doublering") {
785 1 : std::size_t pos = curConfig.find(":");
786 1 : std::string algConfigTmp = curConfig.substr(0, pos + 1) + "ring";
787 1 : algName = g_algName.at(algConfigTmp);
788 1 : return true;
789 1 : }
790 : }
791 9 : auto res = g_algName.find(curConfig);
792 9 : if (res != g_algName.end()) {
793 9 : algName = res->second;
794 9 : return true;
795 : }
796 0 : HCCL_ERROR("[AicpuHcclProcess][%s] algo_name is not exist, algConfig %s is no.", __func__, algConfig.c_str());
797 0 : return false;
798 10 : }
799 :
800 10 : bool SplitHcclAlgoGetLevel1Res(std::string &algoConfig, std::string &algos)
801 : {
802 10 : std::string remainAlgoConfig;
803 10 : std::size_t found = algoConfig.find(";");
804 10 : if ((found == 0) || (found == (algoConfig.length() - 1)) || (found == std::string::npos)) {
805 5 : HCCL_INFO("algoConfig %s thereis no level1 algo config", algoConfig.c_str());
806 5 : return true;
807 : }
808 5 : remainAlgoConfig = algoConfig.substr(found + 1);
809 5 : found = remainAlgoConfig.find(";");
810 5 : std::size_t msgPos = 0;
811 5 : if (found != std::string::npos) {
812 0 : msgPos = found;
813 0 : HCCL_WARNING("[AicpuHcclProcess] algo level is more than 1, not supported !");
814 : } else {
815 5 : msgPos = remainAlgoConfig.size();
816 : }
817 5 : algos = (remainAlgoConfig.substr(0, msgPos));
818 5 : return false;
819 10 : }
820 :
821 5 : HcclResult ParserHcclAlgoLevel1(std::string &algoLevel, uint32_t &level, HcclAlgoType &algoType)
822 : {
823 5 : std::size_t found = algoLevel.find(":");
824 5 : if ((found == 0) || (found == (algoLevel.length() - 1))) {
825 0 : HCCL_ERROR("[Parser][HcclAlgoLevel] algo config is invalid.");
826 0 : return HCCL_E_PARA;
827 : }
828 :
829 5 : std::string orginalLevel = algoLevel.substr(0, found);
830 5 : std::string orginalAlgo = algoLevel.substr(found + 1);
831 :
832 : const std::map<std::string, HcclAlgoType> hcclAlgoTypeMap = {
833 0 : {"null", HcclAlgoType::HCCL_ALGO_TYPE_NULL},
834 0 : {"ring", HcclAlgoType::HCCL_ALGO_TYPE_RING},
835 0 : {"pipeline", HcclAlgoType::HCCL_ALGO_TYPE_PIPELINE},
836 0 : {"fullmesh", HcclAlgoType::HCCL_ALGO_TYPE_FULLMESH},
837 0 : {"H-D_R", HcclAlgoType::HCCL_ALGO_TYPE_HDR},
838 0 : {"pairwise", HcclAlgoType::HCCL_ALGO_TYPE_PAIRWISE},
839 0 : {"NHR", HcclAlgoType::HCCL_ALGO_TYPE_NHR},
840 0 : {"NHR_V1", HcclAlgoType::HCCL_ALGO_TYPE_NHR_V1},
841 0 : {"NB", HcclAlgoType::HCCL_ALGO_TYPE_NB},
842 0 : {"NA", HcclAlgoType::HCCL_ALGO_TYPE_NA},
843 60 : };
844 :
845 5 : auto iterAlgoType = hcclAlgoTypeMap.find(orginalAlgo);
846 5 : if (iterAlgoType == hcclAlgoTypeMap.end()) {
847 0 : HCCL_ERROR("[Parser][HcclAlgoLevel] algo config is invalid, algo %s is not supported.", orginalAlgo.c_str());
848 0 : return HCCL_E_PARA;
849 : }
850 5 : level = HCCL_ALGO_LEVEL_1;
851 5 : algoType = iterAlgoType->second;
852 5 : return HCCL_SUCCESS;
853 10 : }
854 :
855 5 : bool SetAlgTypeLevel1(HcclAlgoType algoConfig, AlgTypeLevel1 &algType, uint32_t moduleNum)
856 : {
857 5 : switch (algoConfig) {
858 0 : case HcclAlgoType::HCCL_ALGO_TYPE_HDR:
859 0 : algType = AlgTypeLevel1::ALG_LEVEL1_HD;
860 0 : break;
861 0 : case HcclAlgoType::HCCL_ALGO_TYPE_RING:
862 0 : algType = AlgTypeLevel1::ALG_LEVEL1_RING;
863 0 : HCCL_INFO("server num[%u]: level1:ring algo is set.", moduleNum);
864 0 : break;
865 0 : case HcclAlgoType::HCCL_ALGO_TYPE_NHR:
866 0 : algType = AlgTypeLevel1::ALG_LEVEL1_NHR;
867 0 : HCCL_INFO("server num[%u]: level1:nhr algo is set.", moduleNum);
868 0 : break;
869 0 : case HcclAlgoType::HCCL_ALGO_TYPE_NHR_V1:
870 0 : algType = AlgTypeLevel1::ALG_LEVEL1_NHR_V1;
871 0 : HCCL_INFO("server num[%u]: level1:nhr_v1 algo is set.", moduleNum);
872 0 : break;
873 0 : case HcclAlgoType::HCCL_ALGO_TYPE_NB:
874 0 : algType = AlgTypeLevel1::ALG_LEVEL1_NB;
875 0 : HCCL_INFO("server num[%u]: level1:nb algo is set.", moduleNum);
876 0 : break;
877 0 : case HcclAlgoType::HCCL_ALGO_TYPE_PIPELINE:
878 0 : algType = AlgTypeLevel1::ALG_LEVEL1_PIPELINE;
879 0 : HCCL_INFO("server num[%u]: level1:pipeline algo is set.", moduleNum);
880 0 : break;
881 5 : case HcclAlgoType::HCCL_ALGO_TYPE_FULLMESH:
882 : case HcclAlgoType::HCCL_ALGO_TYPE_PAIRWISE:
883 5 : HCCL_WARNING("level1:fullmesh algo is not supported. the config is ignored.");
884 : [[fallthrough]];
885 : default:
886 5 : HCCL_WARNING("algo is not supported. the config is ignored.");
887 5 : return false;
888 : }
889 0 : return true;
890 : }
891 :
892 10 : void SetAlgoLevel1(hccl::HcclCommAicpu *commAicpu, HcclAlgoType algoConfig,
893 : uint32_t moduleNum, AlgTypeLevel1 &algType, bool isDefault)
894 : {
895 10 : if ((isDefault == false) && (SetAlgTypeLevel1(algoConfig, algType, moduleNum))) {
896 : // 不使用default配置
897 0 : HCCL_INFO("[AicpuHcclProcess][%s] algType[%u], moduleNum[%u]", __func__, algType, moduleNum);
898 0 : return;
899 : }
900 10 : if (moduleNum >= HCCL_INTER_SERVER_RING_ALGO_MAX_SUPPORT_SERVER_NUM) {
901 : // server 数为 8 以上：使用 HD 算法
902 0 : algType = AlgTypeLevel1::ALG_LEVEL1_HD;
903 : } else {
904 : // server 数为 2 的非整数次幂：使用 RING 算法
905 : // server 数为 2 的整数次幂：使用 HD 算法
906 10 : algType = (((moduleNum & (moduleNum - 1)) != 0) || (moduleNum == 1)) ?
907 : AlgTypeLevel1::ALG_LEVEL1_RING :
908 : AlgTypeLevel1::ALG_LEVEL1_HD;
909 : }
910 10 : DevType devType = commAicpu->GetDevType();
911 10 : if (algType == AlgTypeLevel1::ALG_LEVEL1_HD && devType == DevType::DEV_TYPE_910_93) {
912 10 : algType = AlgTypeLevel1::ALG_LEVEL1_NHR;
913 : }
914 10 : HCCL_INFO("[AicpuHcclProcess][%s] algType[%u], moduleNum[%u]", __func__, algType, moduleNum);
915 : }
916 :
917 10 : void SelectAlgType(hccl::HcclCommAicpu *commAicpu, const std::string &algConfig, uint32_t moduleNum, AlgType &algType)
918 : {
919 : // 当前默认只会穿入0 1两层算法配置，多余层数穿入不做解析.
920 : // 0层算法 当前先写死
921 : // 1层算法 按默认值取
922 10 : AlgTypeLevel0 algType0 = AlgTypeLevel0::ALG_LEVEL0_NP_DOUBLE_RING;
923 : // 构造 1层 algoType, 未填写则取默认值
924 10 : HcclAlgoType level1AlgoConfig = HcclAlgoType::HCCL_ALGO_TYPE_DEFAULT;
925 10 : std::string algos;
926 10 : uint32_t level = 0;
927 10 : AlgTypeLevel1 algType1 = AlgTypeLevel1::ALG_LEVEL1_RESERVED;
928 :
929 10 : std::size_t found = algConfig.find("=");
930 10 : std::string curAlgConfig = algConfig.substr(found + 1);
931 10 : bool useDefault = SplitHcclAlgoGetLevel1Res(curAlgConfig, algos);
932 10 : if (useDefault == false) {
933 5 : ParserHcclAlgoLevel1(algos, level, level1AlgoConfig);
934 : }
935 10 : SetAlgoLevel1(commAicpu, level1AlgoConfig, moduleNum, algType1, useDefault);
936 10 : algType.algoLevel0 = algType0;
937 10 : algType.algoLevel1 = algType1;
938 10 : }
939 :
940 : static const std::unordered_set<std::string> STEP_SIZE_SUPPORT_LIST = {
941 : "AlltoAll=level0:fullmesh;level1:pairwise"
942 : };
943 6 : HcclResult ParseCcOpTilingData(CommonHcclMsg *commonHcclMsg, int32_t groupIdx)
944 : {
945 6 : const HcclTilingVersion version = commonHcclMsg->version;
946 6 : HCCL_INFO("Hccl client message version %u", static_cast<u32>(version));
947 6 : AicpuKfcRpcServerV2 *rpc = GetCommRpcServer(groupIdx);
948 6 : rpc->SetStepSize(0U);
949 6 : rpc->SetTotalStep((0U));
950 6 : if (version == HcclTilingVersion::DEPRECATED_TILING_VERSION) {
951 4 : return HCCL_SUCCESS;
952 : }
953 :
954 2 : Mc2CcTilingInner *mc2CcTiling = reinterpret_cast<Mc2CcTilingInner *>(commonHcclMsg->ccOpTilingData);
955 2 : if (mc2CcTiling == nullptr) {
956 0 : HCCL_ERROR("Tiling is nullptr.");
957 0 : return HCCL_E_PARA;
958 : }
959 :
960 : // 校验tiling的groupName与当前接收数据的group 的index是否一致
961 2 : int32_t tilingGroupIdx = GetComGroupIdx(std::string(mc2CcTiling->groupName));
962 2 : if (tilingGroupIdx != groupIdx) {
963 0 : HCCL_ERROR("Failed to check groupName %s, groupIdx %d, tiling GroupIdx %d",
964 : mc2CcTiling->groupName, groupIdx, tilingGroupIdx);
965 0 : return HCCL_E_PARA;
966 : }
967 :
968 2 : HcclOpResParam *commParam = GetCommAicpuResInst(groupIdx);
969 2 : std::string curAlgName;
970 6 : CHK_PRT_RET(!SelectAlgName(mc2CcTiling->algConfig, commParam->topoInfo.topoType, curAlgName),
971 : HCCL_ERROR("Failed to select algname."), HCCL_E_PARA);
972 2 : AlgType algType;
973 2 : HcclCommAicpu *commAicpu = GetCommAicpuCommInst(groupIdx);
974 4 : SelectAlgType(commAicpu, mc2CcTiling->algConfig, commParam->topoInfo.moduleNum, algType);
975 6 : std::string curTag = std::string(mc2CcTiling->groupName) + std::to_string(mc2CcTiling->opType);
976 2 : SetCommInfoCtx(std::string(mc2CcTiling->groupName), static_cast<u8>(mc2CcTiling->opType),
977 4 : CommInfoCtx{algType, curAlgName, curTag});
978 :
979 2 : if (mc2CcTiling->stepSize > 0U) {
980 0 : CHK_PRT_RET(STEP_SIZE_SUPPORT_LIST.find(mc2CcTiling->algConfig) == STEP_SIZE_SUPPORT_LIST.end(),
981 : HCCL_ERROR("Alg %s is not supported when step size is %u.", mc2CcTiling->algConfig, mc2CcTiling->stepSize),
982 : HCCL_E_PARA);
983 0 : rpc->SetStepSize(mc2CcTiling->stepSize);
984 0 : rpc->SetTotalStep(commParam->rankSize);
985 : }
986 2 : return HCCL_SUCCESS;
987 2 : }
988 :
989 3 : void RepeatUpdateOpParam(hccl::OpParam &opParam, CommonHcclMsg *hcclMsg, HcclMsgExt *hcclMsgExt,
990 : hccl::HcclCommAicpu *commAicpu)
991 : {
992 3 : uint64_t dataLen = hcclMsg->dataCnt * DataUnitSize(hcclMsg->hcclDataType);
993 3 : if (hcclMsg->commType == HcclCMDType::HCCL_CMD_ALLTOALLV || (hcclMsg->commType == HcclCMDType::HCCL_CMD_ALLTOALL && hcclMsg->strideCount > 0)) {
994 10 : for (uint32_t i = 0; i < commAicpu->GetRankSize(); i++) {
995 8 : hcclMsgExt->sendOffset[i] += hcclMsgExt->sendCounts[i];
996 8 : hcclMsgExt->recvOffset[i] += hcclMsgExt->recvCounts[i];
997 : }
998 2 : } else {
999 1 : opParam.outputPtr = reinterpret_cast<void *>(reinterpret_cast<int8_t *>(opParam.outputPtr) + dataLen);
1000 1 : opParam.inputPtr = reinterpret_cast<void *>(reinterpret_cast<int8_t *>(opParam.inputPtr) + dataLen);
1001 : }
1002 3 : }
1003 :
1004 3 : HcclResult AddTaskForHcclMsgV2(hccl::HcclCommAicpu *comm, AicpuKfcRpcServerV2 *rpc, CommonHcclMsg *hcclMsg,
1005 : const HcclOpResParam *commParam)
1006 : {
1007 3 : uint32_t curTurnCntForKernel = 0;
1008 3 : rpc->SetMsgPosForKernel(0);
1009 3 : CommInfoCtx curCtx;
1010 3 : HcclResult ret = GetCommInfoCtx(comm->GetGroupName(), static_cast<uint8_t>(hcclMsg->commType), curCtx);
1011 3 : if (ret != HCCL_SUCCESS) {
1012 0 : HCCL_ERROR("Failed to get comm info from aicpu instance.");
1013 0 : return HCCL_E_INTERNAL;
1014 : }
1015 3 : hccl::OpParam opParam;
1016 3 : std::string algName = curCtx.algName;
1017 3 : opParam.tag = curCtx.tag;
1018 3 : std::string newTag = opParam.tag + "_mc2" + algName + "_device";
1019 :
1020 3 : u32 aicpuAlgType = (static_cast<u32>(curCtx.algType.algoLevel2) << (HCCL_LEVEL_ALGO_WIDTH + HCCL_LEVEL_ALGO_WIDTH)) +
1021 3 : (static_cast<u32>(curCtx.algType.algoLevel1) << HCCL_LEVEL_ALGO_WIDTH) +
1022 3 : static_cast<u32>(curCtx.algType.algoLevel0);
1023 3 : comm->SetAlgType(static_cast<u64>(aicpuAlgType));
1024 3 : PrepareOpParam(&opParam, hcclMsg, *rpc, comm);
1025 : hccl::AlgResourceResponse *algResResponse;
1026 3 : std::unique_ptr<hccl::CollExecutorBase> executor;
1027 6 : while (curTurnCntForKernel < hcclMsg->repeatCnt) {
1028 3 : HCCL_INFO("Orchestrate curTurnCntForKernel %u, hcclMsg->repeatCnt %u", curTurnCntForKernel, hcclMsg->repeatCnt);
1029 3 : curTurnCntForKernel++;
1030 3 : rpc->SetMsgPosForKernel(curTurnCntForKernel);
1031 3 : CHK_RET(comm->GetAlgResponseRes(newTag, algName, opParam, commParam, executor, algResResponse));
1032 3 : HcclResult hcclRet = comm->Orchestrate(newTag, algName, opParam, executor, *algResResponse, commParam);
1033 3 : AicpuKfcProf::GetCurrentAicpuProf()->workCnt++;
1034 3 : CHK_PRT_RET(hcclRet != HCCL_SUCCESS,
1035 : HCCL_ERROR("Executor op fail, opParam.tag[%s], algName[%s]",
1036 : newTag.c_str(), algName.c_str()), hcclRet);
1037 3 : RepeatUpdateOpParam(opParam, hcclMsg, rpc->GetHcclMsgExtPtr(), comm);
1038 : }
1039 3 : return HCCL_SUCCESS;
1040 3 : }
1041 :
1042 15 : HcclResult RunRpcServerLoopProcess(const std::vector<u32> &groupIds, u32 localGroupIdx, bool &finalizeFlag)
1043 : {
1044 : HcclMsg hcclMsg;
1045 : CommonHcclMsg commonHcclMsg;
1046 15 : const u32 groupIdx = groupIds[localGroupIdx];
1047 15 : AicpuKfcRpcServerV2 *rpc = GetCommRpcServer(groupIdx);
1048 15 : HcclCommAicpu *comm = GetCommAicpuCommInst(groupIdx);
1049 15 : HcclOpResParam *commParam = GetCommAicpuResInst(groupIdx);
1050 15 : u32 start = 0U;
1051 15 : u32 end = 0U;
1052 15 : rpc->GetLocalQueueRange(start, end);
1053 15 : const u64 tilingBase = rpc->GetTilingBaseAddr();
1054 : HcclResult ret;
1055 : do {
1056 35 : ret = RpcServerPreCheck(rpc, comm, finalizeFlag);
1057 35 : if (ret == HCCL_E_AGAIN) {
1058 9 : return HCCL_SUCCESS;
1059 26 : } else if (ret != HCCL_SUCCESS) {
1060 2 : return ret;
1061 : }
1062 :
1063 24 : HcclMsg (*msgLists)[HCCL_MSG_CNT] = rpc->GetMsgWorkSpace();
1064 24 : SetMsgEnableFlag(groupIdx, false);
1065 51 : for (u32 i = start; i <= end; ++i) {
1066 28 : if (rpc->GetIsFinalize(i)) {
1067 15 : continue;
1068 : }
1069 :
1070 27 : BarrierStatus status = BarrierStatus::NO_BARRIER;
1071 27 : if (BarrierProcess(groupIdx, localGroupIdx, i, status) != HCCL_SUCCESS) {
1072 1 : rpc->DumpBarrierInfo(localGroupIdx, comm->GetSlaveStream()[i].sqId(), comm->GetDevId());
1073 1 : rpc->PrintAllHcclMsgArea(commParam->rankSize);
1074 1 : return HCCL_E_INTERNAL;
1075 : }
1076 :
1077 26 : if (status != BarrierStatus::NO_BARRIER) {
1078 1 : SetMsgEnableFlag(groupIdx, true);
1079 1 : continue;
1080 : }
1081 :
1082 25 : uint32_t currMsgPos = rpc->GetMsgPos(i);
1083 25 : if (!rpc->ReadAddrMsg(&hcclMsg, msgLists[i], i, currMsgPos, commParam->rankSize)) {
1084 1 : if (rpc->IsExceedLimit(static_cast<HcclCMDType>(hcclMsg.commType.prepareType), commParam->rankSize)) {
1085 0 : return HCCL_E_INTERNAL;
1086 : }
1087 1 : AddMsgInValidCount(groupIdx);
1088 1 : if (GetMsgInValidCount(groupIdx) == LOGCOUNT_PRINT_TIMEOUT) {
1089 0 : HCCL_WARNING("Fail to get msg, addr is %p, queue %u, msgPos %u, group %s",
1090 : msgLists[i], i, currMsgPos, comm->GetGroupName().c_str());
1091 : }
1092 1 : if (rpc->IsPrintLog()) {
1093 0 : LogControl logControl(false, true);
1094 0 : comm->PrintTaskExceptionAllComm();
1095 0 : }
1096 1 : continue;
1097 1 : }
1098 :
1099 24 : if (GetMsgInValidCount(groupIdx) > LOGCOUNT_PRINT_TIMEOUT) {
1100 0 : HCCL_WARNING("Msg channel restores, addr is %p, queue %u, msgPos %u, group %s",
1101 : msgLists[i], i, currMsgPos, comm->GetGroupName().c_str());
1102 : }
1103 24 : SetMsgStartTime(groupIdx);
1104 24 : ClearMsgInValidCount(groupIdx);
1105 24 : SetMsgEnableFlag(groupIdx, true);
1106 :
1107 24 : GetCommonHcclMsg(&hcclMsg, &commonHcclMsg, tilingBase);
1108 24 : HCCL_INFO("Process message queue %u pos %u seq num %u type %u group %s.", i, currMsgPos,
1109 : commonHcclMsg.seqNum, commonHcclMsg.commType, comm->GetGroupName().c_str());
1110 24 : if (commonHcclMsg.commType == HcclCMDType::HCCL_CMD_FINALIZE) {
1111 10 : FinalizeProcess(i, *comm, *rpc);
1112 10 : continue;
1113 14 : } else if (commonHcclMsg.commType == HcclCMDType::HCCL_CMD_INTER_GROUP_SYNC) {
1114 3 : ret = AddTaskForGroupSyncMsg(groupIds, localGroupIdx, &commonHcclMsg);
1115 3 : if (ret == HCCL_E_UNAVAIL) {
1116 2 : SetMsgEnableFlag(groupIdx, false);
1117 2 : rpc->SetNeedRetryFlag(true);
1118 2 : continue;
1119 1 : } else if (ret != HCCL_SUCCESS) {
1120 0 : return ret;
1121 : }
1122 11 : } else if (commonHcclMsg.commType == HcclCMDType::HCCL_CMD_BARRIER) {
1123 2 : rpc->GetBarrierInfoByGroupIdx(localGroupIdx)[i].status = BarrierStatus::SELF_BARRIER;
1124 : } else {
1125 9 : ret = rpc->ProcessExpectPrepareMsg(commonHcclMsg.seqNum, GetExpectPrepareId(i));
1126 9 : if (ret == HCCL_E_UNAVAIL) {
1127 0 : SetMsgEnableFlag(groupIdx, false);
1128 0 : rpc->SetNeedRetryFlag(true);
1129 0 : continue;
1130 9 : } else if (ret != HCCL_SUCCESS) {
1131 0 : return ret;
1132 : }
1133 9 : rpc->SetNeedRetryFlag(false);
1134 9 : rpc->SetMsgRepeatCnt(commonHcclMsg.repeatCnt);
1135 9 : rpc->SetMsgHandlePos(currMsgPos, commonHcclMsg.selfHandleID);
1136 9 : if (commonHcclMsg.commType == HcclCMDType::HCCL_CMD_BATCH_WRITE) {
1137 2 : hccl::OpParam opParam;
1138 2 : PrepareOpParam(&opParam, &commonHcclMsg, *rpc, comm);
1139 2 : CHK_RET(AicpuKfcBatchwriteProcess::BatchWriteProcess(opParam, *comm, *commParam));
1140 2 : } else {
1141 7 : CHK_RET(ParseCcOpTilingData(&commonHcclMsg, groupIdx));
1142 7 : CHK_RET(TaskOrchestrator::IsSupportRDMAReduce(commonHcclMsg.commType, commonHcclMsg.hcclDataType,
1143 : commonHcclMsg.opType));
1144 7 : CHK_RET(AddTaskForHcclMsgV2(comm, rpc, &commonHcclMsg, commParam));
1145 : }
1146 9 : SetExpectPrepareId(i, commonHcclMsg.seqNum + 1U);
1147 : }
1148 12 : rpc->SetMsgPos(i, (currMsgPos + 1) % HCCL_MSG_CNT);
1149 : }
1150 23 : } while (CheckMsgEnableFlag(groupIdx));
1151 3 : return HCCL_SUCCESS;
1152 : }
1153 :
1154 2 : std::string GetNewTag(uint32_t groupIdx)
1155 : {
1156 2 : hccl::HcclCommAicpu *comm = GetCommAicpuCommInst(groupIdx);
1157 2 : AicpuKfcRpcServerV2 *rpc = GetCommRpcServer(groupIdx);
1158 2 : uint32_t currMsgPos = rpc->GetMsgPos();
1159 2 : currMsgPos = currMsgPos > 0 ? currMsgPos - 1 : currMsgPos;
1160 2 : HcclMsg (*msgLists)[HCCL_MSG_CNT] = rpc->GetMsgWorkSpace();
1161 2 : CommInfoCtx curCtx;
1162 2 : GetCommInfoCtx(comm->GetGroupName(), static_cast<HcclCMDType>(msgLists[0U][currMsgPos].commType.prepareType),
1163 : curCtx);
1164 4 : return curCtx.tag + "_mc2" + curCtx.algName + "_device";;
1165 2 : }
1166 :
1167 2 : void ResetRestartParam(RestartParam &restartParam)
1168 : {
1169 2 : restartParam.restartCnt++;
1170 2 : restartParam.restartFlag = false;
1171 2 : restartParam.consultationAllEnd = 0;
1172 8 : for (uint32_t i = 0; i < MAX_COMM_CTX_NUM; i++) {
1173 6 : restartParam.consultationResult[i] = false;
1174 6 : restartParam.linkChanged[i] = false;
1175 6 : restartParam.fsmState[i] = HcclOpExecFSM::HCCL_OP_EXEC_FSM_WAIT_END;
1176 6 : restartParam.errorCode[i] = KfcError::kNone;
1177 : }
1178 2 : }
1179 :
1180 3 : HcclResult RestartProcessConsulation(RestartParam &restartParam, bool &finalizeAllEnd, bool *finalizeMask,
1181 : std::vector<u32> groupIds)
1182 : {
1183 5 : for (size_t i = 0U; i < groupIds.size(); ++i) {
1184 3 : if (restartParam.consultationResult[i]) {
1185 1 : continue;
1186 : }
1187 2 : hccl::HcclCommAicpu *comm = GetCommAicpuCommInst(groupIds[i]);
1188 2 : if (comm == nullptr) {
1189 0 : HCCL_ERROR("Failed to obtain the AICPU communication domain pointer."
1190 : "Check whether the parameters are correct.");
1191 1 : return HCCL_E_PARA;
1192 : }
1193 2 : std::string newTag = GetNewTag(groupIds[i]);
1194 2 : HcclResult ret = AicpuKfcRetryProcess::RetryProcess(*comm, restartParam, i);
1195 2 : if (ret == HCCL_SUCCESS) {
1196 1 : if (restartParam.consultationResult[i]) {
1197 1 : HCCL_RUN_INFO("[MC2][AICPU]MC2 restart process success, groupIdx %u , tag %s", i, newTag.c_str());
1198 1 : restartParam.consultationAllEnd++;
1199 : }
1200 : } else {
1201 : // 重执行协商流程失败，直接返回错误
1202 1 : HCCL_ERROR("[MC2][AICPU]MC2 restart process groupIdx %u failed at state %u ret is %u tag is %s", i, restartParam.fsmState[i], ret, newTag.c_str());
1203 1 : return ret;
1204 : }
1205 2 : }
1206 :
1207 : // 全部协商重执行完成
1208 2 : if (restartParam.consultationAllEnd >= groupIds.size()) {
1209 2 : HCCL_RUN_INFO("[MC2][AICPU]MC2 restart process all group success, reset param and write restart");
1210 2 : SetExpectPrepareId(0U, 0U);
1211 2 : ResetRestartParam(restartParam);
1212 2 : finalizeAllEnd = false;
1213 4 : for (size_t i = 0U; i < groupIds.size(); ++i) {
1214 : // 重置结束标志
1215 2 : finalizeMask[i] = false;
1216 : // 重置rpc
1217 2 : AicpuKfcRpcServerV2 *rpc = GetCommRpcServer(groupIds[i]);
1218 2 : rpc->Reset();
1219 2 : rpc->WriteRestartFlag();
1220 2 : SetMsgStartTime(groupIds[i]);
1221 2 : HCCL_INFO("MC2 restart process reset rpc param end. groupIndex = %u", i);
1222 : }
1223 2 : SetKernelStartTime();
1224 : }
1225 2 : return HCCL_SUCCESS;
1226 : }
1227 :
1228 1 : void RecordReportStatus(const std::vector<u32> &groupIds, dfx::ReportStatus status) {
1229 2 : for (const auto i: groupIds) {
1230 1 : hccl::HcclCommAicpu *comm = GetCommAicpuCommInst(i);
1231 1 : if (comm != nullptr) {
1232 1 : comm->RecordReportStatus(status);
1233 : }
1234 : }
1235 1 : }
1236 :
1237 9 : bool CheckMsgTimeOut(const std::vector<u32> &groupIds) {
1238 9 : if ((GetCurCpuTimestamp() - g_timeOutInfoInst.kernelStartTime) >
1239 : static_cast<unsigned long long>(NSEC_PER_SEC * KERNEL_TIMEOUT)) {
1240 0 : HCCL_ERROR("Kernel Execute TimeOut %lus...", KERNEL_TIMEOUT);
1241 0 : return true;
1242 : }
1243 9 : int timeoutFlag = 0;
1244 24 : for (u32 idx: groupIds) {
1245 15 : if (CheckMsgEnableFlag(idx) && (GetCurCpuTimestamp() - GetMsgStartTime(idx)) >
1246 : static_cast<unsigned long long>(NSEC_PER_SEC * KERNEL_TIMEOUT)) {
1247 0 : HCCL_ERROR("comm group idx %d ReadValidMsg timeout %lus... ", idx, KERNEL_TIMEOUT);
1248 0 : timeoutFlag++;
1249 : }
1250 : }
1251 9 : if (timeoutFlag) {
1252 0 : return true;
1253 : }
1254 9 : return false;
1255 : }
1256 :
1257 17 : HcclResult SetNsOpStatus(const std::vector<u32> &groupIds, bool state)
1258 : {
1259 42 : for (const auto i: groupIds) {
1260 25 : hccl::HcclCommAicpu *comm = GetCommAicpuCommInst(i);
1261 25 : if (comm != nullptr) {
1262 25 : comm->SetNsOpStatus(state);
1263 : }
1264 : }
1265 17 : return HCCL_SUCCESS;
1266 : }
1267 :
1268 12 : HcclResult RunRpcServerInnerProcessV2(const std::vector<u32> &groupIds)
1269 : {
1270 12 : const bool retryEnable = GetOpRetryEnable(groupIds);
1271 12 : RestartParam restartParam;
1272 12 : auto opStartTime = std::chrono::steady_clock::now();
1273 12 : bool finalizeMask[MAX_COMM_CTX_NUM] = {false, false, false};
1274 12 : SetKernelStartTime();
1275 12 : AicpuKfcProf::GetCurrentAicpuProf()->commInitEndTime = GetCurCpuTimestamp(true);
1276 12 : if (CheckNsStopLaunchStatus(groupIds) != HCCL_SUCCESS) {
1277 2 : HCCL_WARNING("the op should not be launched in the suspending status");
1278 2 : return HCCL_E_SUSPENDING;
1279 : }
1280 10 : CHK_RET(SetNsOpStatus(groupIds, true));
1281 : while (true) {
1282 19 : bool finishFlag = true;
1283 45 : for (uint32_t i = 0; i < groupIds.size(); i++) {
1284 29 : if (finalizeMask[i]) {
1285 12 : continue;
1286 : }
1287 17 : finishFlag = false;
1288 17 : if (restartParam.restartFlag) {
1289 0 : continue;
1290 : }
1291 17 : HcclResult res = RunRpcServerLoopProcess(groupIds, i, finalizeMask[i]);
1292 17 : if (res == HCCL_E_SUSPENDING) {
1293 4 : if (retryEnable) {
1294 1 : restartParam.restartFlag = true;
1295 1 : break;
1296 : }
1297 3 : HcclCommAicpu *comm = GetCommAicpuCommInst(groupIds[i]);
1298 3 : if (comm != nullptr && comm->GetNsStopLaunchStatus()) {
1299 2 : finalizeMask[i] = true;
1300 2 : AicpuKfcRpcServerV2 *rpc = GetCommRpcServer(groupIds[i]);
1301 2 : rpc->SetNeedRetryFlag(false);
1302 2 : comm->SetCommRecoveryFlag(true);
1303 2 : (void)comm->BackGroundSetStatus(KfcStatus::kStoplaunch);
1304 : } else {
1305 1 : HCCL_ERROR("[MC2][Restart]Mc2 can not retry, not all comm retryEnable are true");
1306 10 : return res;
1307 : }
1308 13 : } else if (res != HCCL_SUCCESS) {
1309 1 : HCCL_ERROR("RPC server failed to run.");
1310 1 : return res;
1311 : }
1312 : }
1313 :
1314 17 : if (restartParam.restartFlag && HcclAicpuUtils::GetBlockIdx() == 0U) {
1315 1 : HcclResult res = RestartProcessConsulation(restartParam, finishFlag, finalizeMask, groupIds);
1316 1 : if (res != HCCL_SUCCESS) {
1317 1 : HCCL_ERROR("[MC2][AICPU]MC2 restart process failed, restartCnt = %u, res = %u",
1318 : restartParam.restartCnt, res);
1319 1 : RecordReportStatus(groupIds, dfx::ReportStatus::kRetryFail);
1320 1 : return res;
1321 : }
1322 : }
1323 : // 全部结束
1324 16 : if (finishFlag) {
1325 7 : HCCL_INFO("RPC server process ends.");
1326 7 : AicpuKfcProf::GetCurrentAicpuProf()->receiveFinalizeTime = GetCurCpuTimestamp(true);
1327 7 : CHK_RET(SetNsOpStatus(groupIds, false));
1328 7 : if (restartParam.restartCnt > 0) {
1329 0 : auto opEndTime = std::chrono::steady_clock::now();
1330 0 : auto duration = std::chrono::duration_cast<std::chrono::seconds>(opEndTime - opStartTime).count();
1331 0 : HCCL_RUN_INFO("[MC2][AICPU]MC2 restart exec success, restartCnt = %u, take time = %ld s", restartParam.restartCnt, duration);
1332 0 : RecordReportStatus(groupIds, dfx::ReportStatus::kRetrySuccess);
1333 : }
1334 7 : return HCCL_SUCCESS;
1335 : }
1336 : // 消息超时或总执行时间超时
1337 9 : if (CheckMsgTimeOut(groupIds)) {
1338 0 : HCCL_ERROR("RPC server process Timeout.");
1339 0 : for (uint32_t i: groupIds) {
1340 0 : AicpuKfcRpcServerV2 *rpc = GetCommRpcServer(i);
1341 0 : HcclOpResParam *commParam = GetCommAicpuResInst(i);
1342 0 : if (rpc != nullptr && commParam != nullptr) {
1343 0 : rpc->PrintAllHcclMsgArea(commParam->rankSize);
1344 : }
1345 : }
1346 0 : return HCCL_E_TIMEOUT;
1347 : }
1348 9 : }
1349 : return HCCL_SUCCESS;
1350 : }
1351 :
1352 5 : HcclResult RunRpcServerApiV2(void *tilingData, const std::vector<u32> &groupIds)
1353 : {
1354 : // 待适配 startthread DFX
1355 5 : uint32_t commNum = MC2TilingGetHcommCnt(tilingData);
1356 13 : for (uint32_t i = 0; i < commNum; i++) {
1357 8 : Mc2HcommCfg *cfg = MC2TilingGetHcommCfg(tilingData, i);
1358 8 : int32_t groupIdx = GetComGroupIdx(std::string(cfg->groupName));
1359 8 : if (groupIdx < 0) {
1360 0 : HCCL_ERROR("%s idx %d cannot get group by hcomId %s", __func__, i, cfg->groupName);
1361 0 : return HCCL_E_INTERNAL;
1362 : }
1363 8 : hccl::HcclCommAicpu *comm = GetCommAicpuCommInst(groupIdx);
1364 8 : if (comm == nullptr) {
1365 0 : HCCL_ERROR("%s cannot get CommAicpu by groupIdx %d", __func__, groupIdx);
1366 0 : return HCCL_E_INTERNAL;
1367 : }
1368 8 : HcclOpResParam *commParam = GetCommAicpuResInst(groupIdx);
1369 8 : std::string curAlgName;
1370 24 : if (!SelectAlgName(cfg->algConfig, commParam->topoInfo.topoType, curAlgName)) {
1371 0 : return HCCL_E_INTERNAL;
1372 : }
1373 24 : std::string curTag = std::string(cfg->groupName) + std::to_string(cfg->opType);
1374 8 : uint32_t moduleNum = commParam->topoInfo.moduleNum;
1375 8 : AlgType algType;
1376 8 : SelectAlgType(comm, cfg->algConfig, moduleNum, algType);
1377 8 : SetCommInfoCtx(std::string(cfg->groupName), static_cast<u8>(cfg->opType),
1378 16 : CommInfoCtx{algType, curAlgName, curTag});
1379 8 : }
1380 5 : CHK_RET(RunRpcServerInnerProcessV2(groupIds));
1381 3 : return HCCL_SUCCESS;
1382 : }
1383 :
1384 0 : HcclResult KfcStepSizeHandler(const std::vector<u64> &args)
1385 : {
1386 0 : CHK_PRT_RET(args.size() != 3U, HCCL_ERROR("Invalid args size %zu.", args.size()), HCCL_E_INTERNAL);
1387 0 : const AicpuKfcRpcServerV2 *rpc = reinterpret_cast<const AicpuKfcRpcServerV2 *>(args[0]);
1388 0 : u8 stepSize = rpc->GetStepSize();
1389 0 : if (stepSize == 0U) {
1390 0 : HCCL_INFO("The orchestrating OP is not a fine-grained one.");
1391 0 : return HCCL_SUCCESS;
1392 : }
1393 :
1394 0 : Mc2Handler *handler = reinterpret_cast<Mc2Handler *>(args[1]);
1395 0 : handler->version = 0U;
1396 0 : handler->commitAddr = rpc->GetCommitareaAddr(rpc->GetMsgPos());
1397 0 : handler->finishAddr = rpc->GetFinishAddr(rpc->GetMsgPos());
1398 0 : handler->valueAddr = rpc->GetTurnNumAddr();
1399 0 : handler->rankSize = args[2];
1400 0 : handler->repeatCnt = rpc->GetMsgPosForKernel();
1401 0 : handler->stepSize = stepSize;
1402 0 : handler->skipLocalRankCopy = 0U;
1403 0 : handler->skipBufferWindowCopy = 0U;
1404 0 : HCCL_INFO("Prepare MC2 handler: commitAddr %p, finishAddr %p, valueAddr %p, rankSize %u, repeat %u, stepSize %u.",
1405 : handler->commitAddr, handler->finishAddr, handler->valueAddr, handler->rankSize, handler->repeatCnt,
1406 : handler->stepSize);
1407 0 : return HCCL_SUCCESS;
1408 : }
1409 :
1410 0 : HcclResult KfcNotifyPost(const std::vector<u64> &args)
1411 : {
1412 0 : CHK_PRT_RET(args.size() != 3U, HCCL_ERROR("Invalid args size %zu.", args.size()), HCCL_E_INTERNAL);
1413 0 : AicpuKfcRpcServerV2 *rpc = reinterpret_cast<AicpuKfcRpcServerV2 *>(args[0]);
1414 0 : CHK_PRT_RET(rpc == nullptr, HCCL_ERROR("Failed to get rpc pointer."), HCCL_E_INTERNAL);
1415 0 : if (rpc->GetStepSize() != 0 || rpc->GetTotalQueueNum() > 0U) {
1416 0 : HCCL_DEBUG("No need to add notify for MC2.");
1417 0 : return HCCL_SUCCESS;
1418 : }
1419 0 : return rpc->AddCcoreNotify(reinterpret_cast<HcclDispatcher>(args[1]), rpc->GetFinishAddr(rpc->GetMsgPos()),
1420 0 : rpc->GetMsgPosForKernel(), reinterpret_cast<Stream *>(args[2]));
1421 : }
1422 :
1423 0 : HcclResult KfcNotifyWait(const std::vector<u64> &args)
1424 : {
1425 0 : CHK_PRT_RET(args.size() != 3U, HCCL_ERROR("Invalid args size %zu.", args.size()), HCCL_E_INTERNAL);
1426 0 : AicpuKfcRpcServerV2 *rpc = reinterpret_cast<AicpuKfcRpcServerV2 *>(args[0]);
1427 0 : CHK_PRT_RET(rpc == nullptr, HCCL_ERROR("Failed to get rpc pointer."), HCCL_E_INTERNAL);
1428 0 : if (rpc->GetStepSize() != 0 || rpc->GetTotalQueueNum() > 0U) {
1429 0 : HCCL_DEBUG("No need to add wait for MC2.");
1430 0 : return HCCL_SUCCESS;
1431 : }
1432 0 : return rpc->AddCcoreWait(reinterpret_cast<HcclDispatcher>(args[1]), rpc->GetCommitareaAddr(rpc->GetMsgPos()),
1433 0 : rpc->GetMsgPosForKernel(), reinterpret_cast<Stream *>(args[2]), false);
1434 : }
1435 :
1436 0 : HcclResult KfcClearMsgArea(const std::vector<u64> &args)
1437 : {
1438 0 : CHK_PRT_RET(args.size() != 1U, HCCL_ERROR("Invalid args size %zu.", args.size()), HCCL_E_INTERNAL);
1439 0 : AicpuKfcRpcServerV2 *rpc = reinterpret_cast<AicpuKfcRpcServerV2 *>(args[0]);
1440 0 : HcclMsgArea *hcclMsgArea = rpc->GetHcclMsgArea();
1441 0 : if (hcclMsgArea != nullptr) {
1442 0 : (void)memset_s(hcclMsgArea, sizeof(HcclMsgArea), 0, sizeof(HcclMsgArea));
1443 : }
1444 0 : hcclMsgArea->controlMsg.resetSeq = 1;
1445 0 : return HCCL_SUCCESS;
1446 : }
1447 :
1448 0 : HcclResult KfcClearCommitTurn(const std::vector<u64> &args)
1449 : {
1450 0 : CHK_PRT_RET(args.size() != 1U, HCCL_ERROR("Invalid args size %zu.", args.size()), HCCL_E_INTERNAL);
1451 0 : AicpuKfcRpcServerV2 *rpc = reinterpret_cast<AicpuKfcRpcServerV2 *>(args[0]);
1452 0 : HcclMsgArea *hcclMsgArea = rpc->GetHcclMsgArea();
1453 0 : if (hcclMsgArea != nullptr) {
1454 0 : for (uint32_t i = 0; i < HCCL_MSG_CNT; i++) {
1455 0 : hcclMsgArea->commMsg.singleMsg.commitTurnCnt[i].cnt = 0xFF;
1456 : }
1457 : }
1458 0 : return HCCL_SUCCESS;
1459 : }
1460 :
1461 15 : HcclResult PrepareHcommInstance(HcclOpResParam *commParam, const Mc2InitTilingInner *tiling = nullptr)
1462 : {
1463 15 : const std::string &group = commParam->hcomId;
1464 15 : hccl::HcclCommAicpu *hcclCommAicpu = AicpuHcclProcess::AicpuGetCommbyGroup(group);
1465 15 : CHK_PRT_RET(hcclCommAicpu == nullptr,
1466 : HCCL_ERROR("RunAicpuRpcSrvLaunchV2 get Hcclcomm error group [%s]", group.c_str()), HCCL_E_INTERNAL);
1467 :
1468 15 : DevType devType = hcclCommAicpu->GetDevType();
1469 15 : CHK_PRT_RET(devType != DevType::DEV_TYPE_910_93,
1470 : HCCL_ERROR("Platform %u not support, please use 910_93 platform.", static_cast<u32>(devType)),
1471 : HCCL_E_INTERNAL);
1472 :
1473 15 : const DfxExtendInfo *dfxInfo = hcclCommAicpu->GetDfxExtendInfo();
1474 15 : CHK_PRT_RET(dfxInfo->cqeStatus != dfx::CqeStatus::kDefault ||
1475 : dfxInfo->pollStatus == PollStatus::kStopAsException,
1476 : HCCL_ERROR("Exist errors before, cqeStatus:%d, pollStatus:%d, group[%s]",
1477 : dfxInfo->cqeStatus, dfxInfo->pollStatus, group.c_str()), HCCL_E_INTERNAL);
1478 :
1479 15 : const u32 groupIdx = InsertComIdMap(group);
1480 15 : HcclResult ret = InsertCommInst(groupIdx, hcclCommAicpu, commParam);
1481 15 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("Failed to insert comm inst."), HCCL_E_INTERNAL);
1482 :
1483 15 : AicpuKfcRpcServerV2 *rpcServer = GetCommRpcServer(groupIdx);
1484 15 : CHK_PRT_RET(rpcServer == nullptr,
1485 : HCCL_ERROR("RunAicpuRpcSrvLaunchV2 get rpc inst error idx %d group [%s]", groupIdx, group.c_str()),
1486 : HCCL_E_INTERNAL);
1487 :
1488 15 : ret = rpcServer->Init(commParam->mc2WorkSpace, tiling);
1489 15 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("Failed to init for group [%s]", group.c_str()), HCCL_E_INTERNAL);
1490 :
1491 15 : hcclCommAicpu->SetIsDeviceMode(true);
1492 15 : hcclCommAicpu->SetAicpuRpcServer(reinterpret_cast<u64>(rpcServer));
1493 15 : hcclCommAicpu->RegisterKfcHandler(AicpuKfcHandlerType::kSetStepSize, KfcStepSizeHandler);
1494 15 : hcclCommAicpu->RegisterKfcHandler(AicpuKfcHandlerType::kNotifyRecord, KfcNotifyPost);
1495 15 : hcclCommAicpu->RegisterKfcHandler(AicpuKfcHandlerType::kNotifyWait, KfcNotifyWait);
1496 15 : hcclCommAicpu->RegisterKfcHandler(AicpuKfcHandlerType::kClearMsgArea, KfcClearMsgArea);
1497 15 : hcclCommAicpu->RegisterKfcHandler(AicpuKfcHandlerType::kClearCommitTurn, KfcClearCommitTurn);
1498 15 : hcclCommAicpu->RegisterKfcHandler(AicpuKfcHandlerType::kSetProfTimeStart,
1499 0 : [](const std::vector<u64>& args) -> HcclResult {
1500 0 : AicpuKfcProf::SetKfcTimeLine(KfcTimeLine::HCC_EXEC_START_TIME);
1501 0 : return HCCL_SUCCESS;
1502 : });
1503 15 : hcclCommAicpu->RegisterKfcHandler(AicpuKfcHandlerType::kSetProfTimeOrch,
1504 0 : [](const std::vector<u64>& args) -> HcclResult {
1505 0 : AicpuKfcProf::SetKfcTimeLine(KfcTimeLine::SEND_TASK_START_TIME);
1506 0 : return HCCL_SUCCESS;
1507 : });
1508 15 : hcclCommAicpu->RegisterKfcHandler(AicpuKfcHandlerType::kSetProfTimeEnd,
1509 0 : [](const std::vector<u64>& args) -> HcclResult {
1510 0 : AicpuKfcProf::SetKfcTimeLine(KfcTimeLine::SEND_SQE_FINISH_TIME);
1511 0 : return HCCL_SUCCESS;
1512 : });
1513 15 : return HCCL_SUCCESS;
1514 15 : }
1515 : ANONYMOUS_NAMESPACE_END
1516 :
1517 81 : u32 AicpuKfcProcess::AicpuRpcResInit(HccCommResParamTask *commParam)
1518 : {
1519 81 : HcclAicpuUtils::PrintHcclCombinOpParam(*commParam);
1520 :
1521 81 : AicpuComContext *ctx = AicpuGetComContext();
1522 81 : if (ctx->alreadyInit) {
1523 1 : if (strncmp(ctx->hcomId, commParam->hcomId, HCCL_COMM_DOMAIN_KEY_MAX_LEN)) {
1524 1 : HCCL_ERROR("the comm domain is not valid old [%s] != new[%s].", ctx->hcomId, commParam->hcomId);
1525 1 : return AC_ERROR_INVALID_PARAM;
1526 : }
1527 0 : HCCL_INFO("The ctx was already inited");
1528 0 : return 0;
1529 : }
1530 80 : AicpuSqeContext::InitSqeContext();
1531 80 : memset_s(ctx, sizeof(AicpuComContext), 0, sizeof(AicpuComContext));
1532 80 : s32 enableEvent = 0;
1533 80 : ctx->logLevel = dlog_getlevel(HCCL, &enableEvent);
1534 80 : ctx->rankId = commParam->rankId;
1535 80 : ctx->rankNum = commParam->rankNum;
1536 80 : ctx->windowSize = commParam->winSize;
1537 80 : ctx->workSpaceAddr = commParam->mc2WorkSpace.workSpace;
1538 80 : ctx->curTurnCnt = 0;
1539 80 : ctx->commAlg = 0;
1540 80 : ctx->multiServerFlag = commParam->multiServerFlag;
1541 80 : std::iota(ctx->turnValue, ctx->turnValue + TILING_TURN_MAX * AC_MAX_RANK_NUM, 0);
1542 80 : HcclSignalInfo *sigInfo = &commParam->signalInfo.aicpuNotify;
1543 80 : std::shared_ptr<LocalNotify> localNitfy;
1544 80 : EXCEPTION_CATCH((localNitfy = std::make_shared<LocalNotify>()), return HCCL_E_PTR);
1545 80 : CHK_RET(localNitfy->Init(*sigInfo, NotifyLoadType::DEVICE_NOTIFY));
1546 80 : ctx->kfcNotifyId = sigInfo->resId;
1547 :
1548 80 : CHK_RET(hrtDrvGetLocalDevIDByHostDevID(sigInfo->devId, &(ctx->devId)));
1549 :
1550 80 : if (ctx->multiServerFlag) {
1551 0 : CHK_RET(InitIbversData(commParam, ctx));
1552 : } else {
1553 80 : InitRankInfo(commParam, ctx);
1554 80 : CHK_RET(InitSignalInfo(commParam, ctx));
1555 80 : CHK_RET(InitEventId(commParam, ctx));
1556 : }
1557 :
1558 80 : CHK_RET(AicpuKfcProcess::InitStreamInfo(commParam, ctx));
1559 80 : CHK_RET(InitAicpuOpNotify(commParam, ctx));
1560 80 : CHK_RET(InitTimeOutConfig(commParam, ctx));
1561 80 : HCCL_INFO("remote_udevid: %u, local_devid: %u, ssid: %u", sigInfo->devId, ctx->devId, ctx->ssid);
1562 80 : ctx->directlySendMainSteramSqe = false;
1563 80 : ctx->clusterId = HcclAicpuUtils::GetCurClusterId();
1564 80 : auto ret = strcpy_s(ctx->hcomId, sizeof(ctx->hcomId), commParam->hcomId);
1565 80 : HCCL_DEBUG("Init hcom group [%s] strcpy ret %d", ctx->hcomId, static_cast<int>(ret));
1566 80 : ctx->determinism = (commParam->config.deterministic != 0);
1567 80 : ctx->retryEnable = (commParam->config.retryEnable == 1);
1568 80 : ctx->retryHoldTime = commParam->config.retryHoldTime;
1569 80 : ctx->retryIntervalTime = commParam->config.retryIntervalTime;
1570 80 : HCCL_DEBUG("[%s] ctx->retryEnable [%d], ctx->retryHoldTime [%u], ctx->retryIntervalTime [%u]",
1571 : __func__, ctx->retryEnable, ctx->retryHoldTime, ctx->retryIntervalTime);
1572 80 : CHK_RET(InitChipType(ctx));
1573 80 : ctx->overflowAddr = commParam->overFlowAddr;
1574 80 : ctx->onlyRead = commParam->onlyRead;
1575 80 : ctx->dfxExtendInfo.dfxTimeOutConfig.useCredit = true;
1576 80 : dfx::AicpuProfilingManager::Init(ctx);
1577 80 : ctx->alreadyInit = true;
1578 80 : ctx->commOpenStatus = true;
1579 80 : ctx->opIndex = 0;
1580 80 : if (commParam->kfcControlTransferH2DParams.buffLen != 0) {
1581 72 : EXCEPTION_CATCH((ctx->kfcControlTransferH2D = std::make_shared<hccl::HDCommunicate>()), return HCCL_E_PTR);
1582 72 : CHK_SMART_PTR_NULL(ctx->kfcControlTransferH2D);
1583 72 : CHK_RET(ctx->kfcControlTransferH2D->InitDevice(commParam->kfcControlTransferH2DParams));
1584 : }
1585 80 : if (commParam->kfcStatusTransferD2HParams.buffLen != 0) {
1586 72 : EXCEPTION_CATCH((ctx->kfcStatusTransferD2H = std::make_shared<hccl::HDCommunicate>()), return HCCL_E_PTR);
1587 72 : CHK_SMART_PTR_NULL(ctx->kfcStatusTransferD2H);
1588 72 : CHK_RET(ctx->kfcStatusTransferD2H->InitDevice(commParam->kfcStatusTransferD2HParams));
1589 : }
1590 80 : AicpuHcclProcess::CopyCtxInfo(ctx);
1591 80 : AicpuHcclProcess::CallMC2MaintenanceThread(ctx);
1592 80 : if (MC2TraceUtils::Init() != HCCL_SUCCESS) {
1593 0 : HCCL_ERROR("Init trace failed.");
1594 0 : return static_cast<u32>(HCCL_E_INTERNAL);
1595 : }
1596 80 : HCCL_RUN_INFO("End %s", __func__);
1597 80 : return 0;
1598 80 : }
1599 :
1600 : std::unordered_map<int32_t, uint32_t> g_streamIdMap;
1601 8 : u32 AicpuKfcProcess::GetStreamRankIdx(s32 actualStreamId)
1602 : {
1603 8 : auto it = g_streamIdMap.find(actualStreamId);
1604 8 : return it == g_streamIdMap.cend() ? UINT32_MAX : it->second;
1605 : }
1606 :
1607 8 : HcclResult AicpuKfcProcess::DealReturnValue(const AicpuComContext *ctx, const HcclResult ret)
1608 : {
1609 8 : if (ctx->isStopLaunch) {
1610 1 : AicpuHcclProcess::CopyCtxForBackGroundDfx(ctx);
1611 1 : CHK_RET(AicpuHdcUtils::SetOpExecStatus(ctx->kfcStatusTransferD2H, KfcStatus::kStoplaunch, KfcError::kNone, 0));
1612 1 : return HCCL_E_SUSPENDING;
1613 7 : } else if (ctx->endStopLaunch) {
1614 0 : return HCCL_E_SUSPENDING;
1615 : } else {
1616 7 : CHK_RET(AicpuHdcUtils::SetOpExecStatus(ctx->kfcStatusTransferD2H, KfcStatus::kError, KfcError::kInner, 0));
1617 7 : return ret;
1618 : }
1619 : }
1620 :
1621 20 : HcclResult AicpuKfcProcess::AddTaskForHcclMsg(AicpuComContext *ctx, AicpuKfcRpcServer &rpc, CommonHcclMsg *hcclMsg,
1622 : AivAicpuOpParam *msg, u64 tilingBase)
1623 : {
1624 : // reduce scatter:在strideLen使能的情况下，如果recvCount * repeat > strideLen 则偏移越界，报错
1625 20 : if (hcclMsg->commType == HcclCMDType::HCCL_CMD_REDUCE_SCATTER && hcclMsg->strideCount != 0 &&
1626 0 : hcclMsg->dataCnt * hcclMsg->repeatCnt > hcclMsg->strideCount) {
1627 0 : HCCL_ERROR("In ReduceScatter algorithm, when stride Count is not zero, repeatCnt * dataCnt"
1628 : " should not be greater than strideCount.");
1629 0 : hcclMsg->PrintMsg("");
1630 0 : return HCCL_E_PARA;
1631 : }
1632 :
1633 20 : AivAicpuOpParam *tmpptr = nullptr;
1634 20 : AivAicpuOpParam nextMsg;
1635 20 : u64 dataLen = DataUnitSize(msg->hcclDataType) * msg->count;
1636 20 : ctx->curTurnCntForKernel = 0;
1637 20 : ctx->totalTurnCntForKernel = hcclMsg->repeatCnt;
1638 43 : while (ctx->curTurnCntForKernel < hcclMsg->repeatCnt) {
1639 24 : HCCL_INFO("ctx->curTurnCntForKernel %u, hcclMsg->repeatCnt %u", ctx->curTurnCntForKernel,
1640 : hcclMsg->repeatCnt);
1641 : // 当前msg预取仅支持当前及下一条msg都为allgather
1642 24 : if (hcclMsg->commType == HcclCMDType::HCCL_CMD_ALLGATHER &&
1643 8 : (hcclMsg->hcclDataType == HCCL_DATA_TYPE_FP16 || hcclMsg->hcclDataType == HCCL_DATA_TYPE_BFP16)) {
1644 8 : HCCL_INFO("Try get AllGather next msg");
1645 : HcclMsg tmpMsg;
1646 8 : if (ctx->curTurnCntForKernel < (hcclMsg->repeatCnt - 1)) {
1647 1 : GetNextMsgFromMsg(msg, &nextMsg, dataLen, ctx->rankNum);
1648 1 : tmpptr = &nextMsg;
1649 7 : } else if (rpc.CheckRcvAddrMsg(&tmpMsg, ctx->msgPosForKernel + 1)) {
1650 : CommonHcclMsg commonHcclMsg;
1651 7 : GetCommonHcclMsg(&tmpMsg, &commonHcclMsg, tilingBase);
1652 7 : rpc.HcclMsg2AicAicpuOpParam(&commonHcclMsg, &nextMsg);
1653 7 : tmpptr = &nextMsg;
1654 : } else {
1655 0 : HCCL_INFO("nextMsg is not ready. msgPos %u", ctx->msgPosForKernel + 1);
1656 0 : tmpptr = nullptr;
1657 : }
1658 : // 如果nextMsg和hcclMsg不同commtype或datatype，nextMsg要置为nullptr
1659 8 : if (tmpptr != nullptr && (tmpptr->commType != hcclMsg->commType ||
1660 2 : tmpptr->hcclDataType != hcclMsg->hcclDataType)) {
1661 6 : HCCL_INFO("Set nextMsg nullptr");
1662 6 : tmpptr = nullptr;
1663 : }
1664 : }
1665 24 : ctx->curTurnCntForKernel++;
1666 24 : CHK_RET(AicpuKfcProcess::AicpuCcOpExe(msg, tmpptr, ctx));
1667 23 : TaskOrchestrator::ActiveRecordMain(AicpuKfcProcess::GetActiveSqId(ctx));
1668 : // 更新msg
1669 23 : UpdateMsg(msg, dataLen, ctx->rankNum);
1670 : }
1671 19 : return HCCL_SUCCESS;
1672 : }
1673 :
1674 25 : HcclResult AicpuKfcProcess::RunRpcServerApi(AicpuComContext *ctx, AicpuKfcRpcServer &rpc, u64 tilingBase)
1675 : {
1676 25 : if (ctx->devType != DevType::DEV_TYPE_910B) {
1677 1 : HCCL_ERROR("Platform not support, please use 910B platform.");
1678 1 : return HCCL_E_PARA;
1679 : }
1680 : HcclMsg hcclMsg;
1681 : CommonHcclMsg commonHcclMsg;
1682 24 : AivAicpuOpParam msg;
1683 24 : AicpuUpdatComContextMumber(offsetof(AicpuComContext, dfxExtendInfo.kfcStatus), DfxKfcStatus::kOneStart);
1684 24 : AicpuHcclProcess::CallMC2MaintenanceThread(ctx);
1685 24 : ctx->directlySendMainSteramSqe = true;
1686 24 : ctx->msgPosForKernel = 0;
1687 :
1688 24 : msg.opId.index = ctx->opIndex + 1;
1689 24 : AicpuUpdatComContextMumber(offsetof(AicpuComContext, opIndex), msg.opId.index);
1690 24 : if (ctx->endStopLaunch) {
1691 0 : HCCL_WARNING("the op should not be launched in suspending status");
1692 0 : return HCCL_E_SUSPENDING;
1693 : }
1694 24 : CHK_RET(AicpuHdcUtils::InitOpExecStatus(ctx->kfcStatusTransferD2H, msg.opId));
1695 24 : AicpuUpdatComContextMumber(offsetof(AicpuComContext, isOpLaunch), true);
1696 : while (true) {
1697 45 : HCCL_INFO("start to read the [%u] msg", ctx->msgPosForKernel);
1698 45 : if (!rpc.ReadAddrMsg(&hcclMsg, ctx->msgPosForKernel)) {
1699 5 : HCCL_ERROR("fail to get addr msg, msgPos %u", ctx->msgPosForKernel);
1700 5 : rpc.PrintAllHcclMsgArea();
1701 5 : TaskOrchestrator::PrintTimeOutSqInfo(ctx, ctx->dfxExtendInfo.dfxTimeOutConfig.sqeWaitTimeOut);
1702 5 : return HCCL_E_TIMEOUT;
1703 : }
1704 40 : GetCommonHcclMsg(&hcclMsg, &commonHcclMsg, tilingBase);
1705 : // 处理finalzie消息
1706 40 : if (commonHcclMsg.commType == HcclCMDType::HCCL_CMD_FINALIZE) {
1707 17 : AicpuKfcProf::GetProInst(*ctx).receiveFinalizeTime = GetCurCpuTimestamp(true);
1708 17 : if (ctx->debugMode == MC2_DEBUG_PRINT_BUFF) {
1709 1 : rpc.PrintAllHcclMsgAreaData();
1710 : }
1711 17 : break;
1712 23 : } else if (commonHcclMsg.commType == HcclCMDType::HCCL_CMD_INIT) {
1713 0 : continue;
1714 23 : } else if (commonHcclMsg.commType == HcclCMDType::HCCL_CMD_INTER_GROUP_SYNC ||
1715 23 : commonHcclMsg.commType == HcclCMDType::HCCL_CMD_BARRIER) {
1716 0 : HCCL_ERROR("Msg %u is not supported.", static_cast<uint32_t>(commonHcclMsg.commType));
1717 0 : return HCCL_E_PARA;
1718 23 : } else if (commonHcclMsg.commType == HcclCMDType::HCCL_CMD_BATCH_WRITE) {
1719 : // 校验多机场景，multiServerFlag必须为true
1720 2 : if (!ctx->multiServerFlag) {
1721 0 : HCCL_ERROR("Batch write is only support in multi server.");
1722 0 : return HCCL_E_PARA;
1723 : }
1724 : // 处理BatchWrite的操作从直接发送->队列发送。
1725 2 : CHK_RET(AicpuKfcBatchwriteProcess::HandleBatchWriteOperation(commonHcclMsg, ctx));
1726 : // 刷一下标记内存 commitTUrnCnt=0, finsihTurnCnt++
1727 2 : rpc.WriteTurnCnt(ctx->msgPosForKernel);
1728 : } else {
1729 21 : rpc.HcclMsg2AicAicpuOpParam(&commonHcclMsg, &msg);
1730 21 : if (msg.sendBuffer == 0UL || msg.recvBuffer == 0UL) {
1731 1 : HCCL_ERROR("Get msg buffer is nullptr.");
1732 1 : msg.PrintMsg("Invalid msg buffer");
1733 1 : rpc.PrintAllHcclMsgArea();
1734 1 : return HCCL_E_PARA;
1735 : }
1736 20 : CHK_RET(SetMsgWinOffset(ctx, &msg));
1737 20 : CHK_RET(AicpuKfcProcess::AddTaskForHcclMsg(ctx, rpc, &commonHcclMsg, &msg, tilingBase));
1738 : }
1739 : // 切换到下一个msg
1740 21 : ctx->msgPosForKernel = (ctx->msgPosForKernel + 1) % HCCL_MSG_CNT;
1741 21 : }
1742 : // 添加结束任务
1743 17 : if (!ctx->multiServerFlag) {
1744 15 : CHK_RET(AicpuDispatcher::AddAllEndTaskOnMainStream(AicpuKfcProcess::GetActiveSqId(ctx)));
1745 15 : TaskOrchestrator::ActiveRecordMain(AicpuKfcProcess::GetActiveSqId(ctx));
1746 15 : ctx->directlySendMainSteramSqe = false;
1747 15 : CHK_RET(AicpuKfcProcess::WaitTaskFinish(ctx));
1748 : } else {
1749 2 : AicpuKfcBatchwriteProcess::FinishProcess();
1750 : }
1751 17 : rpc.WriteFinishWhenAllFinalize(ctx->msgPosForKernel);
1752 17 : AicpuUpdatComContextMumber(offsetof(AicpuComContext, dfxExtendInfo.kfcStatus), DfxKfcStatus::kOneFinished);
1753 17 : return HCCL_SUCCESS;
1754 : }
1755 :
1756 3 : HcclResult AicpuKfcProcess::AicpuRunRpcServerForApi(AicpuComContext *ctx, u64 tilingBase) {
1757 3 : static AicpuKfcRpcServer rpc;
1758 3 : rpc.Init(ctx->workSpaceAddr);
1759 3 : AicpuKfcProf::GetProInst(*ctx).commInitEndTime = GetCurCpuTimestamp(true);
1760 3 : const HcclResult ret = RunRpcServerApi(ctx, rpc, tilingBase);
1761 3 : AicpuUpdatComContextMumber(offsetof(AicpuComContext, isOpLaunch), false);
1762 3 : if (ret != HCCL_SUCCESS) {
1763 0 : return DealReturnValue(ctx, ret);
1764 : } else {
1765 3 : CHK_RET(AicpuHdcUtils::SetOpExecStatus(ctx->kfcStatusTransferD2H, KfcStatus::kEnd, KfcError::kNone, 0));
1766 3 : return ret;
1767 : }
1768 : }
1769 :
1770 8 : u32 AicpuKfcProcess::AicpuRunRpcServerForMC2V2(KFCTaskV2 *task, const Mc2InitTilingInner *tilingData)
1771 : {
1772 : static std::atomic<bool> initFlag(false);
1773 8 : if (HcclAicpuUtils::GetBlockNum() <= 1U || !initFlag.exchange(true)) {
1774 17 : for (u64 i = 0UL; i < task->ctxNum; i++) {
1775 9 : HcclOpResParam *ctx = reinterpret_cast<HcclOpResParam *>(task->context[i]);
1776 9 : HcclAicpuUtils::PrintHcclOpResParam(ctx);
1777 9 : CHK_PRT_RET(PrepareHcommInstance(ctx, tilingData) != HCCL_SUCCESS,
1778 : AicpuHcclProcess::AicpuReleaseCommbyGroup(ctx->hcomId),
1779 : HCCL_E_INTERNAL);
1780 : }
1781 : }
1782 8 : CHK_PRT_RET(AicpuKfcUtils::ThreadBarrier(BARRIER_TIMEOUT) != HCCL_SUCCESS,
1783 : HCCL_ERROR("[%s]Timeout during instance preparation.", __func__),
1784 : HCCL_E_INTERNAL);
1785 :
1786 8 : std::vector<u32> groupIds{};
1787 17 : for (u64 i = 0UL; i < task->ctxNum; i++) {
1788 9 : HcclOpResParam *ctx = reinterpret_cast<HcclOpResParam *>(task->context[i]);
1789 18 : groupIds.emplace_back(GetComGroupIdx(ctx->hcomId));
1790 : }
1791 8 : HcclResult ret = RunRpcServerInnerProcessV2(groupIds);
1792 8 : CHK_PRT_RET(AicpuKfcUtils::ThreadBarrier(BARRIER_TIMEOUT) != HCCL_SUCCESS,
1793 : HCCL_ERROR("[%s]Timeout during instance finalize.", __func__),
1794 : HCCL_E_INTERNAL);
1795 :
1796 8 : if (HcclAicpuUtils::GetBlockIdx() == 0U) {
1797 17 : for (u64 i = 0UL; i < task->ctxNum; i++) {
1798 9 : HcclOpResParam *ctx = reinterpret_cast<HcclOpResParam *>(task->context[i]);
1799 18 : AicpuHcclProcess::AicpuReleaseCommbyGroup(ctx->hcomId);
1800 : }
1801 8 : initFlag = false;
1802 8 : if (CheckNsStopLaunchStatus(groupIds) == HCCL_E_SUSPENDING) {
1803 1 : SetExpectPrepareId(0U, 0U);
1804 1 : HCCL_INFO("mc2 opp is suspended");
1805 1 : return AICPUSUSPENDING_ERROR;
1806 : }
1807 : }
1808 7 : return ret;
1809 8 : }
1810 :
1811 5 : u32 AicpuKfcProcess::AicpuRunRpcServerForMC2(KFCTaskV2 *task)
1812 : {
1813 5 : HcclOpResParam *commParam[MAX_COMM_CTX_NUM]{};
1814 5 : std::vector<u32> groupIds{};
1815 13 : for (int i = 0; i < static_cast<int>(task->ctxNum); i++) {
1816 8 : commParam[i] = reinterpret_cast<HcclOpResParam *>(task->context[i]);
1817 8 : CHK_RET(static_cast<HcclResult>(PrepareHcommInstance(commParam[i])));
1818 16 : groupIds.emplace_back(GetComGroupIdx(commParam[i]->hcomId));
1819 : }
1820 5 : HcclResult ret = RunRpcServerApiV2(reinterpret_cast<void *>(task->tilingData), groupIds);
1821 13 : for (int i = 0; i < static_cast<int>(task->ctxNum); i++) {
1822 8 : std::string group = commParam[i]->hcomId;
1823 8 : AicpuHcclProcess::AicpuReleaseCommbyGroup(group);
1824 8 : }
1825 5 : if (CheckNsStopLaunchStatus(groupIds) == HCCL_E_SUSPENDING) {
1826 2 : SetExpectPrepareId(0U, 0U);
1827 2 : HCCL_INFO("mc2 opp is suspended");
1828 2 : return AICPUSUSPENDING_ERROR;
1829 : }
1830 3 : return ret;
1831 5 : }
1832 :
1833 95 : HcclResult AicpuKfcProcess::AicpuCcOpExe(AivAicpuOpParam *commParam, AivAicpuOpParam *commParamNext,
1834 : AicpuComContext *ctx)
1835 : {
1836 95 : HCCL_DEBUG("----------start %s -------", __func__);
1837 95 : if (commParam == nullptr || ctx == nullptr) {
1838 0 : HCCL_ERROR("%s commParam or ctx is null.", __func__);
1839 0 : return HCCL_E_PARA;
1840 : }
1841 :
1842 : // 1. process global resource, update context.
1843 95 : ctx->unitSize = DataUnitSize(commParam->hcclDataType);
1844 95 : CHK_PRT_RET(ctx->unitSize == 0, HCCL_ERROR("[%s]ctx->unitSize is zero.", __func__), HCCL_E_PARA);
1845 95 : ctx->commLen = ctx->unitSize * commParam->count;
1846 95 : ctx->commType = commParam->commType;
1847 95 : ctx->reducekind = commParam->opType;
1848 95 : ctx->commOpType = GetCcOpType(ctx->commLen, ctx->rankNum); // twoshot.onshot...
1849 95 : ctx->totalTurnCnt = commParam->totalTurnCnt;
1850 95 : ctx->useBufferType = commParam->useBufferType;
1851 95 : ctx->winOffset = commParam->winOffset;
1852 :
1853 95 : auto profInst = AicpuKfcProf::GetProInst(*ctx);
1854 95 : if (AicpuKfcUtils::NeedRecordTimeTaken(*ctx)) {
1855 10 : u32 index = profInst.workCnt;
1856 10 : index = (index >= AC_MAX_PROF_COMM_CNT) ? (AC_MAX_PROF_COMM_CNT - 1) : index;
1857 10 : profInst.commLoop[index].dataLen = ctx->commLen;
1858 : }
1859 :
1860 95 : HcclResult result = TaskOrchestrator::RunConcreteAlgorithm(commParam, commParamNext, ctx);
1861 95 : if (result != HCCL_SUCCESS) {
1862 3 : HCCL_ERROR("Run comm alg failed, rankId:%d, result:%u.", ctx->rankId, result);
1863 3 : return result;
1864 : }
1865 92 : profInst.workCnt = ctx->curTurnCnt;
1866 : // 所有轮次执行完毕后通知aclnn
1867 92 : if (ctx->curTurnCnt == ctx->totalTurnCnt &&
1868 34 : (ctx->devType != DevType::DEV_TYPE_310P1 && ctx->devType != DevType::DEV_TYPE_310P3) &&
1869 34 : ctx->preparePosition != TASK_PREPARE_KERNEL) {
1870 34 : CHK_RET(AicpuDispatcher::AddAllEndTaskOnMainStream(AicpuKfcProcess::GetActiveSqId(ctx)));
1871 : }
1872 :
1873 92 : return HCCL_SUCCESS;
1874 : }
1875 :
1876 51 : HcclResult AicpuKfcProcess::WaitTaskFinish(AicpuComContext *ctx, bool isWaitTask)
1877 : {
1878 51 : HcclResult ret = HCCL_SUCCESS;
1879 51 : CHK_RET(AicpuKfcUtils::TraceProfSubmit());
1880 51 : if (isWaitTask || ctx->retryEnable) {
1881 51 : ret = TaskOrchestrator::WaitMainStreamFinish(ctx);
1882 51 : CHK_PRT_RET((ret != HCCL_SUCCESS && ret != HCCL_E_SUSPENDING),
1883 : HCCL_ERROR("wait main stream finish failed"), ret);
1884 : }
1885 49 : return ret;
1886 : }
1887 :
1888 83 : HcclResult AicpuKfcProcess::ResetSqBuff(AicpuComContext *ctx)
1889 : {
1890 83 : CHK_RET(AicpuSqeContext::ClearLocalBuff());
1891 83 : SqeContext *sqeContext = GetSqeContext();
1892 83 : u32 streamNum = (ctx->multiServerFlag) ? 1 : ctx->rankNum;
1893 747 : for (u32 i = 0; i < streamNum; i++) {
1894 664 : auto &buff = sqeContext->buffPtr[i];
1895 664 : CHK_RET(QuerySqStatusByType(ctx->devId, ctx->streamInfo[i].sqId, DRV_SQCQ_PROP_SQ_TAIL, buff.sqTail));
1896 664 : CHK_RET(QuerySqStatusByType(ctx->devId, ctx->streamInfo[i].sqId, DRV_SQCQ_PROP_SQ_HEAD, buff.sqHead));
1897 664 : HCCL_INFO("hccl aicpu reset stream buffer, sqid:%d head:%u tail:%u.",
1898 : ctx->streamInfo[i].sqId, buff.sqHead, buff.sqTail);
1899 : }
1900 83 : HCCL_INFO("reset stream sq buffer success.");
1901 83 : return HCCL_SUCCESS;
1902 : }
1903 :
1904 156 : u32 AicpuKfcProcess::GetActiveSqId(AicpuComContext *ctx)
1905 : {
1906 156 : return ctx->rankId;
1907 : }
1908 :
1909 80 : HcclResult AicpuKfcProcess::InitStreamInfo(HccCommResParamTask *commParam, AicpuComContext *ctx)
1910 : {
1911 80 : g_streamIdMap.clear();
1912 80 : u32 streamNum = (ctx->multiServerFlag) ? 1U : ctx->rankNum;
1913 720 : for (u32 i = 0; i < streamNum; i++) {
1914 640 : auto &streamInfo = ctx->streamInfo[i];
1915 640 : streamInfo.sqId = commParam->streamInfo[i].sqIds;
1916 640 : streamInfo.logicCqId = commParam->streamInfo[i].logicCqids;
1917 640 : streamInfo.actualStreamId = commParam->streamInfo[i].streamIds;
1918 640 : HCCL_INFO("streamInfo.sqId :%d, streamId:%d", streamInfo.sqId, streamInfo.actualStreamId);
1919 640 : u64 sq_addr = 0;
1920 640 : CHK_RET(QuerySqBaseAddr(ctx->devId, streamInfo.sqId, sq_addr));
1921 640 : streamInfo.sqBaseAddr = reinterpret_cast<void *>(sq_addr);
1922 640 : CHK_RET(QuerySqStatusByType(ctx->devId, streamInfo.sqId, DRV_SQCQ_PROP_SQ_DEPTH, streamInfo.sqDepth));
1923 640 : g_streamIdMap[streamInfo.actualStreamId] = i;
1924 : }
1925 80 : CHK_RET(AicpuKfcProcess::ResetSqBuff(ctx));
1926 80 : return HCCL_SUCCESS;
1927 : }
|
