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