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_rpc_server.h"
12 :
13 : #include "log_control.h"
14 : #include "hccl_tiling_msg.h"
15 : #include "algorithm/task_orchestrator.h"
16 : #include "common/aicpu_hccl_common.h"
17 : #include "common/aicpu_kfc_utils.h"
18 : #include "utils/aicpu_hdc_utils.h"
19 :
20 : using namespace HcclApi;
21 62 : void AicpuKfcRpcServer::Init(u64 workSpaceAddr, uint32_t notifyOff, uint16_t notifyBeginCnt, KFCTask* taskParam)
22 : {
23 62 : tilingData_ = reinterpret_cast<HcclKFCTilingData*>(taskParam->tilingData);
24 :
25 : // 为提升效率，workspace 必须512 对齐
26 62 : u64 addr = workSpaceAddr;
27 62 : HCCL_DEBUG("AicpuKfcRpcServer::Init addr:%u", addr);
28 :
29 : // 规划每个AIV的消息接收地址， 总计使用： MAX_AIV_NUM * HCCL_MSG_CNT
30 62 : if (static_cast<TASK_PREPARE_POSITION>(tilingData_->preparePosition) == TASK_PREPARE_HOST) {
31 39 : msgBody_ = reinterpret_cast<RpcMsgBody*>(addr);
32 39 : msgBody_->msgRcvArea[0][0].res[0] = 0U;
33 39 : msgSndWorkArea_ = reinterpret_cast<AivAicpuOpParam*>(workSpaceAddr + notifyOff);
34 : msgRcvRspArea_
35 39 : = reinterpret_cast<AivAicpuOpParam*>(workSpaceAddr + notifyOff + notifyBeginCnt * sizeof(u8) * AC_SQE_SIZE);
36 2535 : for (uint32_t i = 0; i < AC_MAX_AIV; i++) {
37 2496 : rcvMsgPos_[i] = 0;
38 2496 : sndMsgPos_[i] = 0;
39 2496 : aivState_[i] = 0;
40 : }
41 : } else {
42 23 : hcclMsgArea_ = reinterpret_cast<HcclMsgArea*>(addr);
43 : }
44 :
45 62 : genTaskNum_ = 0;
46 62 : genTaskParam_ = taskParam;
47 62 : }
48 :
49 3 : void AicpuKfcRpcServer::Init(u64 workSpaceAddr)
50 : {
51 3 : hcclMsgArea_ = reinterpret_cast<HcclMsgArea*>(workSpaceAddr);
52 3 : genTaskNum_ = 0;
53 3 : genTaskParam_ = nullptr;
54 3 : tilingData_ = nullptr;
55 3 : }
56 :
57 0 : bool AicpuKfcRpcServer::PostMsg(uint32_t curTurnCnt) const
58 : {
59 0 : AivAicpuOpParam* msg = msgRcvRspArea_;
60 0 : msg->rcvCnt = curTurnCnt;
61 0 : msg->valid = HCCL_MSG_VALID_MASK;
62 0 : msg->PrintMsg("Snd");
63 :
64 : #ifdef __aarch64__
65 : __asm__ __volatile__("dsb st" : : : "memory");
66 : #endif
67 :
68 0 : return true;
69 : }
70 :
71 17 : void AicpuKfcRpcServer::WriteFinishWhenAllFinalize(uint32_t msgPos)
72 : {
73 17 : hcclMsgArea_->commMsg.singleMsg.finishedTurnCnt[msgPos].cnt = FINALIZE_FINISH_CNT; // 用于校验的非法值
74 17 : HCCL_INFO(
75 : "Post finishedTurnCnt[%u].cnt = %lu.", msgPos, hcclMsgArea_->commMsg.singleMsg.finishedTurnCnt[msgPos].cnt);
76 : #ifdef __aarch64__
77 : __asm__ __volatile__("dsb st" : : : "memory");
78 : #endif
79 17 : }
80 :
81 2 : void AicpuKfcRpcServer::WriteTurnCnt(uint32_t msgPos)
82 : {
83 2 : hcclMsgArea_->commMsg.singleMsg.commitTurnCnt[msgPos].cnt = 0;
84 2 : hcclMsgArea_->commMsg.singleMsg.finishedTurnCnt[msgPos].cnt += 1;
85 2 : HCCL_INFO(
86 : "Post position %u commitTurnCnt cnt = %lu, finishedTurnCnt cnt = %lu.", msgPos,
87 : hcclMsgArea_->commMsg.singleMsg.commitTurnCnt[msgPos].cnt,
88 : hcclMsgArea_->commMsg.singleMsg.finishedTurnCnt[msgPos].cnt);
89 : #ifdef __aarch64__
90 : __asm__ __volatile__("dsb st" : : : "memory");
91 : #endif
92 2 : }
93 :
94 3085 : inline std::string AicpuKfcRpcServer::GetMsgTypeString(uint8_t msgType)
95 : {
96 3085 : if (msgType == RANK_ADDR) {
97 6162 : return "Addr";
98 : }
99 4 : if (msgType == RANK_WORK) {
100 2 : return "work";
101 : }
102 3 : if (msgType == RANK_ADD_AND_WORK) {
103 2 : return "Addr&work";
104 : }
105 2 : if (msgType == RANK_TAIL_TIME) {
106 2 : return "EndNotify";
107 : }
108 2 : return "unknown";
109 : }
110 :
111 : #pragma GCC push_options
112 : #pragma GCC optimize("O0")
113 0 : bool AicpuKfcRpcServer::RcvMsg(AivAicpuOpParam* rMsg, uint32_t aivID, uint8_t msgType)
114 : {
115 0 : if (rMsg == nullptr) {
116 0 : return false;
117 : }
118 0 : auto pos = rcvMsgPos_[aivID];
119 0 : auto msg = &msgBody_->msgRcvArea[aivID][pos];
120 0 : if (NeedAutoGenMsg()) {
121 0 : HCCL_DEBUG("RcvMsg by task param:%d/%d", genTaskNum_ + 1, tilingData_->turnNum);
122 0 : GenMsgByTaskParam(rMsg);
123 0 : genTaskNum_++;
124 : } else {
125 0 : HCCL_DEBUG("RcvMsg on msg:%p, aivId:%d, pos:%d", msg, aivID, pos);
126 :
127 : #ifdef __aarch64__
128 : __asm__ __volatile__("dsb ld" : : : "memory");
129 : #endif
130 : #ifdef __amd64__
131 0 : __asm__ __volatile__("" : : : "memory");
132 : #endif
133 :
134 : do {
135 0 : } while (!ReadValidMsg(rMsg, msg, msgType, false));
136 : }
137 :
138 0 : msg->PrintMsg(GetMsgTypeString(msgType));
139 :
140 0 : if (rMsg->isLast) {
141 0 : aivState_[aivID] = 1;
142 : }
143 :
144 0 : pos = (pos + 1) % HCCL_MSG_CNT;
145 0 : rcvMsgPos_[aivID] = pos;
146 :
147 0 : return true;
148 : }
149 :
150 : template <typename T>
151 1 : bool AicpuKfcRpcServer::ReadValidMsg(T* rMsg, T* msg, uint8_t msgType, bool reset)
152 : {
153 : (void)msgType;
154 1 : if (msg->valid != HCCL_MSG_VALID_MASK) {
155 1 : return false;
156 : }
157 0 : *rMsg = *msg;
158 0 : if (reset) {
159 0 : msg->valid = ~HCCL_MSG_VALID_MASK;
160 : }
161 : #ifdef __aarch64__
162 : __asm__ __volatile__("dsb st" : : : "memory");
163 : #endif
164 0 : HCCL_INFO("reset valid value %u", msg->valid);
165 0 : return true;
166 : }
167 :
168 49 : bool AicpuKfcRpcServer::CheckDebugMode(HcclMsg* rMsg)
169 : {
170 49 : auto ctx = AicpuGetComContext();
171 49 : if ((ctx->debugMode == MC2_DEBUG_PREPARE_TIMEOUT)
172 1 : && (rMsg->commType.msgType != ControlMsgType::HCCL_CMD_FINALIZE)) {
173 1 : return false;
174 : }
175 :
176 48 : if ((ctx->debugMode == MC2_DEBUG_FINALIZE_TIMEOUT)
177 2 : && (rMsg->commType.msgType == ControlMsgType::HCCL_CMD_FINALIZE)) {
178 1 : return false;
179 : }
180 47 : return true;
181 : }
182 :
183 30098917 : bool AicpuKfcRpcServer::ReadApiValidMsg(HcclMsg* rMsg, HcclMsg* msg, bool reset)
184 : {
185 : #ifdef __aarch64__
186 : __asm__ __volatile__("dsb ld" : : : "memory");
187 : #endif
188 : #ifdef __amd64__
189 30098917 : __asm__ __volatile__("" : : : "memory");
190 : #endif
191 30098917 : if (msg->addMsg.v0Msg.valid != HCCL_MSG_VALID_MASK) {
192 30098867 : CHK_RET(AicpuKfcUtils::TraceProfSubmit());
193 30098867 : return false;
194 : }
195 50 : memcpy_s(rMsg, sizeof(HcclMsg), msg, sizeof(HcclMsg));
196 50 : uint32_t modifiedXor = AicpuKfcUtils::GenXor(rMsg);
197 : static uint32_t xorCheckNum = 0;
198 50 : if (xorCheckNum % MC2_API_XORCHECK_PRINT_NUM == 0 && modifiedXor != rMsg->addMsg.v0Msg.xorCheck) {
199 1 : HCCL_RUN_INFO(
200 : "[MC2] data is modified! rMsg:%s msg:%s, modifiedXor:%u, origin_xor:%u.",
201 : AicpuKfcUtils::GetMsgSimpleStr(*rMsg).c_str(), AicpuKfcUtils::GetMsgSimpleStr(*msg).c_str(), modifiedXor,
202 : rMsg->addMsg.v0Msg.xorCheck);
203 1 : xorCheckNum++;
204 1 : return false;
205 : }
206 : #ifdef __aarch64__
207 : __asm__ __volatile__("dsb ld" : : : "memory");
208 : #endif
209 : #ifdef __amd64__
210 49 : __asm__ __volatile__("" : : : "memory");
211 : #endif
212 : static uint32_t cmpCheckNum = 0;
213 49 : if (memcmp(rMsg, msg, sizeof(HcclMsg)) != 0) {
214 0 : if (cmpCheckNum % MC2_API_XORCHECK_PRINT_NUM == 0) {
215 0 : HCCL_RUN_INFO(
216 : "[MC2] Check msg equal fail, rMsg:%s msg:%s", AicpuKfcUtils::GetMsgSimpleStr(*rMsg).c_str(),
217 : AicpuKfcUtils::GetMsgSimpleStr(*msg).c_str());
218 : }
219 0 : cmpCheckNum++;
220 0 : return false;
221 : }
222 :
223 49 : if (reset) {
224 42 : msg->addMsg.v0Msg.valid = ~HCCL_MSG_VALID_MASK;
225 : }
226 :
227 49 : if (!CheckDebugMode(rMsg)) {
228 2 : return false;
229 : }
230 : #ifdef __aarch64__
231 : __asm__ __volatile__("dsb st" : : : "memory");
232 : #endif
233 47 : HCCL_INFO("reset valid value %u", msg->addMsg.v0Msg.valid);
234 47 : return true;
235 : }
236 :
237 : #pragma GCC pop_options
238 :
239 34 : bool AicpuKfcRpcServer::ReadAddrMsg(AivAicpuOpParam* rMsg, uint32_t aivID)
240 : {
241 : (void)aivID;
242 34 : GenMsgByTaskParam(rMsg);
243 34 : return true;
244 : }
245 :
246 1 : bool AicpuKfcRpcServer::ReadWorkMsg(AivAicpuOpParam* rMsg, uint32_t aivID, uint32_t curTurnCnt)
247 : {
248 : (void)aivID;
249 1 : return ReadValidMsg(rMsg, msgSndWorkArea_, RANK_WORK, false) && (curTurnCnt <= rMsg->sendCnt);
250 : }
251 :
252 0 : bool AicpuKfcRpcServer::CheckRcvWorkMsg(AivAicpuOpParam* rMsg, uint32_t aivID, uint32_t curTurnCnt)
253 : {
254 : (void)aivID;
255 : #ifdef __aarch64__
256 : __asm__ __volatile__("dsb ld" : : : "memory");
257 : #endif
258 : #ifdef __amd64__
259 0 : __asm__ __volatile__("" : : : "memory");
260 : #endif
261 0 : HCCL_INFO("CheckRcvWorkMsg, curTurnCnt %u", curTurnCnt);
262 0 : rMsg->PrintMsg(GetMsgTypeString(RANK_MSG_TYPE::RANK_WORK));
263 0 : uint32_t loopCnt = 0;
264 0 : u64 startUsec = GetCurCpuTimestamp();
265 : do {
266 : /************调测使用，正式交付的时候删除************/
267 0 : if (loopCnt > 10000) { // 10000 is max loop cnt
268 0 : loopCnt = 0;
269 : // 打印所有流的sq状态
270 0 : HCCL_INFO(
271 : "current states %s Msg %p[sendCnt:%d, valid:%d, curTurnCnt %u",
272 : GetMsgTypeString(RANK_MSG_TYPE::RANK_WORK).c_str(), msgSndWorkArea_, msgSndWorkArea_->sendCnt,
273 : msgSndWorkArea_->valid, curTurnCnt);
274 : }
275 :
276 0 : if (GetCurCpuTimestamp() - startUsec > static_cast<unsigned long long>(NSEC_PER_SEC) * 6) { // 6 is over time
277 0 : HCCL_ERROR("ReadValidMsg timeout 6s... ");
278 0 : break;
279 : }
280 0 : loopCnt++;
281 : /************************************************/
282 0 : } while (!(ReadValidMsg(rMsg, msgSndWorkArea_, RANK_MSG_TYPE::RANK_WORK, false) && (curTurnCnt <= rMsg->sendCnt)));
283 :
284 0 : rMsg->PrintMsg(GetMsgTypeString(RANK_MSG_TYPE::RANK_WORK));
285 0 : return true;
286 : }
287 :
288 72 : bool AicpuKfcRpcServer::CheckRcvAddrMsg(AivAicpuOpParam* rMsg, uint32_t aivID)
289 : {
290 72 : HCCL_INFO("RcvMsg by task param %u", tilingData_->turnNum);
291 72 : GenMsgByTaskParam(rMsg);
292 72 : genTaskNum_++;
293 :
294 72 : if (rMsg->isLast != 0) {
295 37 : aivState_[aivID] = 1;
296 : }
297 72 : HCCL_INFO("CheckRcvAddrMsg, genTaskNum %u", genTaskNum_);
298 72 : rMsg->PrintMsg(GetMsgTypeString(RANK_MSG_TYPE::RANK_ADDR));
299 :
300 72 : return true;
301 : }
302 :
303 7 : bool AicpuKfcRpcServer::CheckRcvAddrMsg(HcclMsg* hcclMsg, uint32_t msgPos)
304 : {
305 7 : if (!ReadApiValidMsg(hcclMsg, &(hcclMsgArea_->commMsg.singleMsg.sendMsgs[msgPos]), false)) {
306 0 : return false;
307 : }
308 7 : AicpuKfcUtils::PrintMsg("CheckRcvAddrMsg hcclMsg", *hcclMsg);
309 7 : return true;
310 : }
311 :
312 45 : bool AicpuKfcRpcServer::ReadAddrMsg(HcclMsg* hcclMsg, uint32_t msgPos)
313 : {
314 45 : auto ctx = AicpuGetComContext();
315 45 : if (ctx == nullptr) {
316 0 : HCCL_ERROR("Get ctx is nullptr");
317 0 : return false;
318 : }
319 45 : uint32_t loopCnt = 0;
320 45 : u64 startUsec = GetCurCpuTimestamp();
321 : #ifdef CCL_LLT
322 45 : const u64 warningThreshold = static_cast<unsigned long long>(NSEC_PER_SEC);
323 45 : const u64 errorThreshold = static_cast<unsigned long long>(NSEC_PER_SEC);
324 : #else
325 : const u64 warningThreshold = static_cast<unsigned long long>(NSEC_PER_SEC) * MC2_API_MSG_TIMEOUT;
326 : const u64 errorThreshold = static_cast<unsigned long long>(NSEC_PER_SEC) * dfx::kKfcTimeOut;
327 : #endif
328 45 : u8 eventPrintTurn = 1; // 标记 Event日志的打印
329 : do {
330 30098914 : if (ctx->dfxExtendInfo.pollStatus == PollStatus::kStopAsException) {
331 1 : HCCL_ERROR("hccl aicpu exec failed, for exception.");
332 5 : return false;
333 : }
334 :
335 30098913 : KfcCommand cmd = KfcCommand::kNone;
336 30098913 : CHK_RET(AicpuHdcUtils::GetOpExecCtrlCmd(ctx->kfcControlTransferH2D, cmd));
337 30098913 : if ((cmd == KfcCommand::NsStopLaunch) && (ctx->commOpenStatus) && (!ctx->endStopLaunch)) {
338 1 : HCCL_WARNING("N second stop Launch for recv stop launch cmd.");
339 1 : AicpuUpdatComContextMumber(offsetof(AicpuComContext, isStopLaunch), true);
340 1 : AicpuUpdatComContextMumber(offsetof(AicpuComContext, endStopLaunch), true);
341 1 : return false;
342 : }
343 30098912 : if (loopCnt > 10000) { // 10000 is max loop cnt
344 3008 : loopCnt = 0;
345 : // 打印所有流的sq状态
346 3008 : HCCL_INFO(
347 : "current states %s Msg %p, msgPos %u", GetMsgTypeString(RANK_MSG_TYPE::RANK_ADDR).c_str(),
348 : &(hcclMsgArea_->commMsg.singleMsg.sendMsgs[msgPos]), msgPos);
349 : }
350 30098912 : const u64 passedTs = GetCurCpuTimestamp() - startUsec;
351 30098912 : if (passedTs > warningThreshold * eventPrintTurn) {
352 3 : HCCL_RUN_WARNING(
353 : "[AicpuKfcRpcServer][ReadAddrMsg] ReadValidMsg[%u] timeout %lus", msgPos,
354 : warningThreshold / static_cast<unsigned long long>(NSEC_PER_SEC));
355 3 : LogControl logControl(false, true);
356 3 : PrintAllHcclMsgArea();
357 3 : if (!ctx->multiServerFlag) {
358 3 : TaskOrchestrator::PrintTimeOutSqInfo(
359 : ctx, warningThreshold / static_cast<unsigned long long>(NSEC_PER_SEC));
360 : }
361 3 : eventPrintTurn *= 2; // 2 is print event log times
362 3 : if (passedTs > errorThreshold) {
363 3 : return false;
364 : }
365 3 : }
366 30098909 : loopCnt++;
367 30098909 : } while (!(ReadApiValidMsg(hcclMsg, &(hcclMsgArea_->commMsg.singleMsg.sendMsgs[msgPos]), true)));
368 :
369 : // 打印读消息的时间
370 40 : if (eventPrintTurn > 1) {
371 0 : HCCL_RUN_INFO(
372 : "[AicpuKfcRpcServer][ReadAddrMsg] Read HcclMsg[%u] cost[%llu ns]", msgPos,
373 : GetCurCpuTimestamp() - startUsec);
374 : } else {
375 40 : HCCL_INFO(
376 : "[AicpuKfcRpcServer][ReadAddrMsg] Read HcclMsg[%u] cost[%llu ns]", msgPos,
377 : GetCurCpuTimestamp() - startUsec);
378 : }
379 :
380 40 : PrintMsg(hcclMsg, msgPos);
381 40 : return true;
382 : }
383 :
384 28 : void AicpuKfcRpcServer::HcclMsg2AicAicpuOpParam(CommonHcclMsg* hcclMsg, AivAicpuOpParam* opMsg)
385 : {
386 28 : HcclApi::Mc2CcTilingInner* innerTiling = reinterpret_cast<HcclApi::Mc2CcTilingInner*>(hcclMsg->ccOpTilingData);
387 28 : AicpuComContext* ctx = AicpuGetComContext();
388 28 : if (tilingData_ == nullptr && innerTiling == nullptr) {
389 0 : HCCL_ERROR("Invalid tiling data, please check opType or other fields.");
390 0 : return;
391 : }
392 28 : opMsg->commType = hcclMsg->commType;
393 28 : opMsg->opType = hcclMsg->opType;
394 28 : opMsg->sendBuffer = hcclMsg->sendBuffer;
395 28 : opMsg->recvBuffer = hcclMsg->recvBuffer;
396 28 : opMsg->winOffset = 0U;
397 28 : opMsg->count = hcclMsg->commType == HcclCMDType::HCCL_CMD_REDUCE_SCATTER ? hcclMsg->dataCnt * ctx->rankNum :
398 : hcclMsg->dataCnt;
399 28 : opMsg->hcclDataType = hcclMsg->hcclDataType;
400 28 : opMsg->isLast = 0U;
401 28 : opMsg->sendCnt = 0x34;
402 28 : opMsg->rcvCnt = 0x12;
403 28 : opMsg->valid = hcclMsg->valid;
404 28 : opMsg->everyTurnRsp = hcclMsg->everyTurnRsp;
405 28 : opMsg->everyTurnWait = hcclMsg->everyTurnWait;
406 28 : opMsg->strideLen = static_cast<u64>(hcclMsg->strideCount);
407 :
408 28 : if (tilingData_ != nullptr) {
409 28 : opMsg->funID = tilingData_->funID;
410 28 : opMsg->totalTurnCnt = tilingData_->turnNum;
411 28 : opMsg->useBufferType = tilingData_->useBufferType;
412 : } else {
413 0 : opMsg->useBufferType = innerTiling->skipBufferWindowCopy;
414 0 : ctx->skipLocalDataCopy = innerTiling->skipLocalRankCopy;
415 : }
416 28 : if (ctx->gatherOut == 0U && opMsg->commType == HcclCMDType::HCCL_CMD_ALLGATHER) {
417 0 : ctx->gatherOut = opMsg->recvBuffer;
418 : }
419 28 : HCCL_DEBUG(
420 : "useBufferType:%u, recvBuffer[%#llx], gatherOut[%#llx], commType[%d].", opMsg->useBufferType, opMsg->recvBuffer,
421 : ctx->gatherOut, opMsg->commType);
422 : // 不需要gather out，就不需要拷贝本卡数据。需要gather out时，如果是aic负责拷贝本卡数据，reduceOp设为1即可
423 28 : if (opMsg->commType == HcclCMDType::HCCL_CMD_ALLGATHER && opMsg->opType != HCCL_REDUCE_PROD) {
424 9 : opMsg->opType = ctx->skipLocalDataCopy ? HCCL_REDUCE_PROD : HCCL_REDUCE_SUM;
425 : }
426 84 : opMsg->PrintMsg("CheckRcvAddrMsg opMsg");
427 : }
428 :
429 109 : bool AicpuKfcRpcServer::CheckAivIsEnd(uint32_t aivId) { return (aivState_[aivId] == 1); }
430 :
431 0 : bool AicpuKfcRpcServer::NeedAutoGenMsg() { return genTaskParam_ != nullptr && genTaskNum_ < tilingData_->turnNum; }
432 :
433 106 : bool AicpuKfcRpcServer::GenMsgIsLastMsg() { return (genTaskNum_ + 1 == tilingData_->turnNum); }
434 :
435 0 : uint8_t AicpuKfcRpcServer::GetWaitPolicy() { return (tilingData_->waitPolicy); }
436 :
437 37 : uint8_t AicpuKfcRpcServer::GetTaskType() const { return (tilingData_->taskType); }
438 :
439 0 : uint8_t AicpuKfcRpcServer::GetRspPolicy() { return (tilingData_->rspPolicy); }
440 :
441 0 : uint8_t AicpuKfcRpcServer::GetGenTaskNum() { return genTaskNum_; }
442 :
443 60 : TASK_PREPARE_POSITION AicpuKfcRpcServer::GetPreparePosition() const
444 : {
445 60 : return static_cast<TASK_PREPARE_POSITION>(tilingData_->preparePosition);
446 : }
447 :
448 106 : void AicpuKfcRpcServer::GenMsgByTaskParam(AivAicpuOpParam* outMsg)
449 : {
450 106 : outMsg->commType = static_cast<HcclCMDType>(tilingData_->commType);
451 106 : outMsg->opType = static_cast<HcclReduceOp>(tilingData_->reduceOp);
452 :
453 106 : switch (outMsg->commType) {
454 42 : case HcclCMDType::HCCL_CMD_ALLGATHER: {
455 42 : CalcAllgatherBuffer(outMsg);
456 :
457 : // 不需要gather out，就不需要拷贝本卡数据。需要gather out时，如果是aic负责拷贝本卡数据，reduceOp设为1即可
458 42 : if (outMsg->opType != HCCL_REDUCE_PROD) {
459 42 : outMsg->opType = tilingData_->hasCommOut ? HCCL_REDUCE_SUM : HCCL_REDUCE_PROD;
460 : }
461 42 : break;
462 : }
463 33 : case HcclCMDType::HCCL_CMD_ALLREDUCE: {
464 33 : CalcAllreduceBuffer(outMsg);
465 33 : break;
466 : }
467 28 : case HcclCMDType::HCCL_CMD_REDUCE_SCATTER: {
468 28 : CalcReduceScatterBuffer(outMsg);
469 28 : break;
470 : }
471 3 : default: {
472 3 : HCCL_ERROR("commType [%d] is not supported.", outMsg->commType);
473 3 : break;
474 : }
475 : }
476 :
477 : outMsg->count
478 106 : = genTaskNum_ < tilingData_->turnNum - tilingData_->tailNum ? tilingData_->sendCnt : tilingData_->tailSendCnt;
479 106 : outMsg->hcclDataType = static_cast<HcclDataType>(tilingData_->dataType);
480 :
481 106 : outMsg->isLast = GenMsgIsLastMsg() ? 1 : 0;
482 106 : outMsg->funID = tilingData_->funID;
483 106 : outMsg->totalTurnCnt = tilingData_->turnNum;
484 106 : outMsg->sendCnt = 0x34;
485 106 : outMsg->rcvCnt = 0x12;
486 106 : outMsg->valid = HCCL_MSG_VALID_MASK;
487 106 : outMsg->everyTurnRsp = tilingData_->rspPolicy;
488 106 : outMsg->everyTurnWait = tilingData_->waitPolicy;
489 106 : outMsg->strideLen = static_cast<u64>(tilingData_->stride);
490 106 : outMsg->useBufferType = tilingData_->useBufferType;
491 106 : }
492 :
493 105 : u64 AicpuKfcRpcServer::GetSendOff() const
494 : {
495 105 : if (tilingData_->commAlg == COMM_ALG_DOUBLE_RING || tilingData_->commAlg == COMM_ALG_SWITCH_WING) {
496 21 : return 0UL;
497 : }
498 84 : const u64 headNum = tilingData_->turnNum - tilingData_->tailNum;
499 84 : if (genTaskNum_ <= headNum) {
500 83 : return genTaskNum_ * tilingData_->sendOff;
501 : }
502 1 : return headNum * tilingData_->sendOff + (genTaskNum_ - headNum) * tilingData_->tailSendOff;
503 : }
504 :
505 105 : u64 AicpuKfcRpcServer::GetRecvOff() const
506 : {
507 105 : if (tilingData_->commAlg == COMM_ALG_DOUBLE_RING || tilingData_->commAlg == COMM_ALG_SWITCH_WING) {
508 21 : return 0UL;
509 : }
510 84 : const u64 headNum = tilingData_->turnNum - tilingData_->tailNum;
511 84 : if (genTaskNum_ <= headNum) {
512 83 : return genTaskNum_ * tilingData_->recvOff;
513 : }
514 1 : return headNum * tilingData_->recvOff + (genTaskNum_ - headNum) * tilingData_->tailRecvOff;
515 : }
516 :
517 42 : void AicpuKfcRpcServer::CalcAllgatherBuffer(AivAicpuOpParam* outMsg) const
518 : {
519 42 : const auto recvOff = GetRecvOff();
520 42 : outMsg->sendBuffer = genTaskParam_->inputA + GetSendOff();
521 42 : if (!tilingData_->useBufferType) {
522 42 : outMsg->recvBuffer = genTaskParam_->commOut + recvOff;
523 : } else {
524 0 : outMsg->recvBuffer = genTaskParam_->workSpace + tilingData_->workspaceOff + recvOff;
525 : }
526 42 : }
527 :
528 33 : void AicpuKfcRpcServer::CalcAllreduceBuffer(AivAicpuOpParam* outMsg) const
529 : {
530 33 : const auto sendOff = GetSendOff();
531 33 : const auto recvOff = GetRecvOff();
532 33 : u64 sendBuffer = 0UL;
533 33 : u64 recvBuffer = 0UL;
534 33 : if (tilingData_->commOrder == 0) { // 通信在前 或 aicpu通信展开(单allreduce)
535 21 : sendBuffer = genTaskParam_->inputA + sendOff;
536 21 : if (!tilingData_->useBufferType) {
537 15 : recvBuffer = genTaskParam_->commOut + recvOff;
538 : } else {
539 6 : recvBuffer = genTaskParam_->workSpace + tilingData_->workspaceOff + recvOff;
540 : }
541 : } else {
542 12 : sendBuffer = genTaskParam_->outputC + sendOff;
543 12 : recvBuffer = genTaskParam_->outputC + recvOff;
544 : }
545 :
546 33 : outMsg->sendBuffer = sendBuffer;
547 33 : outMsg->recvBuffer = recvBuffer;
548 33 : outMsg->winOffset = sendOff;
549 33 : }
550 :
551 28 : void AicpuKfcRpcServer::CalcReduceScatterBuffer(AivAicpuOpParam* outMsg) const
552 : {
553 28 : const auto sendOff = GetSendOff();
554 28 : const auto recvOff = GetRecvOff();
555 28 : u64 sendBuffer = 0UL;
556 28 : u64 recvBuffer = 0UL;
557 28 : if (tilingData_->commOrder == 0) { // aicpu通信展开(单reducescatter)
558 7 : sendBuffer = genTaskParam_->inputA + sendOff;
559 7 : if (!tilingData_->useBufferType) {
560 1 : recvBuffer = genTaskParam_->commOut + recvOff;
561 : } else {
562 6 : recvBuffer = genTaskParam_->workSpace + tilingData_->workspaceOff + recvOff;
563 : }
564 : } else {
565 21 : sendBuffer = genTaskParam_->workSpace + tilingData_->workspaceOff + sendOff;
566 21 : recvBuffer = genTaskParam_->outputC + recvOff;
567 : }
568 28 : outMsg->sendBuffer = sendBuffer;
569 28 : outMsg->recvBuffer = recvBuffer;
570 28 : }
571 :
572 0 : void AicpuKfcRpcServer::ClearWorkMsg() const
573 : {
574 0 : msgSndWorkArea_->sendCnt = 0;
575 0 : msgSndWorkArea_->valid = 0;
576 0 : }
577 :
578 9 : void AicpuKfcRpcServer::PrintAllHcclMsgArea()
579 : {
580 9 : const auto ctx = AicpuGetComContext();
581 9 : if (ctx == nullptr) {
582 0 : return;
583 : }
584 9 : AicpuKfcUtils::PrintAllHcclMsgArea(hcclMsgArea_, ctx->rankNum, true);
585 : }
586 :
587 40 : void AicpuKfcRpcServer::PrintMsg(HcclMsg* hcclMsg, uint32_t msgPos)
588 : {
589 40 : const auto ctx = AicpuGetComContext();
590 40 : if (ctx->debugMode == MC2_DEBUG_PRINT_MSG) {
591 2 : AicpuKfcUtils::PrintMsg("ReadAddrMsg msgPos " + std::to_string(msgPos), *hcclMsg, true);
592 2 : AicpuKfcUtils::PrintAllHcclMsgArea(hcclMsgArea_, ctx->rankNum);
593 : } else {
594 38 : AicpuKfcUtils::PrintMsg("ReadAddrMsg msgPos " + std::to_string(msgPos), *hcclMsg);
595 : }
596 :
597 40 : if (ctx->debugMode == MC2_DEBUG_PRINT_BUFF) {
598 2 : AicpuKfcUtils::PrintApiBufferByMsgPos(*hcclMsg, msgPos);
599 : }
600 40 : }
601 :
602 1 : void AicpuKfcRpcServer::PrintAllHcclMsgAreaData()
603 : {
604 65 : for (uint32_t i = 0; i < HCCL_MSG_CNT; ++i) {
605 64 : AicpuKfcUtils::PrintApiBufferByMsgPos(hcclMsgArea_->commMsg.singleMsg.sendMsgs[i], i);
606 : }
607 1 : }
|
