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 3207 : inline std::string AicpuKfcRpcServer::GetMsgTypeString(uint8_t msgType)
95 : {
96 3207 : if (msgType == RANK_ADDR) {
97 6406 : 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 50 : bool AicpuKfcRpcServer::CopyAndCheckApiMsg(HcclMsg* rMsg, HcclMsg* msg)
184 : {
185 50 : if (memcpy_s(rMsg, sizeof(HcclMsg), msg, sizeof(HcclMsg)) != EOK) {
186 0 : HCCL_ERROR("rMsg memcpy_s failed!,size:%d", sizeof(HcclMsg));
187 0 : return false;
188 : }
189 50 : uint32_t modifiedXor = AicpuKfcUtils::GenXor(rMsg);
190 : static uint32_t xorCheckNum = 0;
191 50 : if (xorCheckNum % MC2_API_XORCHECK_PRINT_NUM == 0 && modifiedXor != rMsg->addMsg.v0Msg.xorCheck) {
192 1 : HCCL_RUN_INFO(
193 : "[MC2] data is modified! rMsg:%s msg:%s, modifiedXor:%u, origin_xor:%u.",
194 : AicpuKfcUtils::GetMsgSimpleStr(*rMsg).c_str(), AicpuKfcUtils::GetMsgSimpleStr(*msg).c_str(), modifiedXor,
195 : rMsg->addMsg.v0Msg.xorCheck);
196 1 : xorCheckNum++;
197 1 : return false;
198 : }
199 : #if defined(__aarch64__)
200 : __asm__ __volatile__("dsb ld" : : : "memory");
201 : #elif defined(__amd64__)
202 49 : __asm__ __volatile__("" : : : "memory");
203 : #endif
204 : static uint32_t cmpCheckNum = 0;
205 49 : if (memcmp(rMsg, msg, sizeof(HcclMsg)) != 0) {
206 0 : if (cmpCheckNum % MC2_API_XORCHECK_PRINT_NUM == 0) {
207 0 : HCCL_WARNING(
208 : "[MC2] Check msg equal failed, rMsg:%s msg:%s", AicpuKfcUtils::GetMsgSimpleStr(*rMsg).c_str(),
209 : AicpuKfcUtils::GetMsgSimpleStr(*msg).c_str());
210 : }
211 0 : cmpCheckNum++;
212 0 : return false;
213 : }
214 :
215 49 : return true;
216 : }
217 :
218 31313345 : bool AicpuKfcRpcServer::ReadApiValidMsg(HcclMsg* rMsg, HcclMsg* msg, bool reset)
219 : {
220 : #if defined(__aarch64__)
221 : __asm__ __volatile__("dsb ld" : : : "memory");
222 : #elif defined(__amd64__)
223 31313345 : __asm__ __volatile__("" : : : "memory");
224 : #endif
225 31313345 : if (msg->addMsg.v0Msg.valid != HCCL_MSG_VALID_MASK) {
226 31313295 : CHK_RET(AicpuKfcUtils::TraceProfSubmit());
227 31313295 : return false;
228 : }
229 50 : if (!CopyAndCheckApiMsg(rMsg, msg)) {
230 1 : return false;
231 : }
232 49 : if (reset) {
233 42 : msg->addMsg.v0Msg.valid = ~HCCL_MSG_VALID_MASK;
234 : }
235 49 : if (!CheckDebugMode(rMsg)) {
236 2 : return false;
237 : }
238 : #ifdef __aarch64__
239 : __asm__ __volatile__("dsb st" : : : "memory");
240 : #endif
241 47 : HCCL_INFO("reset valid value %u", msg->addMsg.v0Msg.valid);
242 47 : return true;
243 : }
244 :
245 : #pragma GCC pop_options
246 :
247 34 : bool AicpuKfcRpcServer::ReadAddrMsg(AivAicpuOpParam* rMsg, uint32_t aivID)
248 : {
249 : (void)aivID;
250 34 : GenMsgByTaskParam(rMsg);
251 34 : return true;
252 : }
253 :
254 1 : bool AicpuKfcRpcServer::ReadWorkMsg(AivAicpuOpParam* rMsg, uint32_t aivID, uint32_t curTurnCnt)
255 : {
256 : (void)aivID;
257 1 : return ReadValidMsg(rMsg, msgSndWorkArea_, RANK_WORK, false) && (curTurnCnt <= rMsg->sendCnt);
258 : }
259 :
260 0 : bool AicpuKfcRpcServer::CheckRcvWorkMsg(AivAicpuOpParam* rMsg, uint32_t aivID, uint32_t curTurnCnt)
261 : {
262 : (void)aivID;
263 : #ifdef __aarch64__
264 : __asm__ __volatile__("dsb ld" : : : "memory");
265 : #endif
266 : #ifdef __amd64__
267 0 : __asm__ __volatile__("" : : : "memory");
268 : #endif
269 0 : HCCL_INFO("CheckRcvWorkMsg, curTurnCnt %u", curTurnCnt);
270 0 : rMsg->PrintMsg(GetMsgTypeString(RANK_MSG_TYPE::RANK_WORK));
271 0 : uint32_t loopCnt = 0;
272 0 : u64 startUsec = GetCurCpuTimestamp();
273 : do {
274 : /************调测使用，正式交付的时候删除************/
275 0 : if (loopCnt > 10000) { // 10000 is max loop cnt
276 0 : loopCnt = 0;
277 : // 打印所有流的sq状态
278 0 : HCCL_INFO(
279 : "current states %s Msg %p[sendCnt:%d, valid:%d, curTurnCnt %u",
280 : GetMsgTypeString(RANK_MSG_TYPE::RANK_WORK).c_str(), msgSndWorkArea_, msgSndWorkArea_->sendCnt,
281 : msgSndWorkArea_->valid, curTurnCnt);
282 : }
283 :
284 0 : if (GetCurCpuTimestamp() - startUsec > static_cast<unsigned long long>(NSEC_PER_SEC) * 6) { // 6 is over time
285 0 : HCCL_ERROR("ReadValidMsg timeout 6s... ");
286 0 : break;
287 : }
288 0 : loopCnt++;
289 : /************************************************/
290 0 : } while (!(ReadValidMsg(rMsg, msgSndWorkArea_, RANK_MSG_TYPE::RANK_WORK, false) && (curTurnCnt <= rMsg->sendCnt)));
291 :
292 0 : rMsg->PrintMsg(GetMsgTypeString(RANK_MSG_TYPE::RANK_WORK));
293 0 : return true;
294 : }
295 :
296 72 : bool AicpuKfcRpcServer::CheckRcvAddrMsg(AivAicpuOpParam* rMsg, uint32_t aivID)
297 : {
298 72 : HCCL_INFO("RcvMsg by task param %u", tilingData_->turnNum);
299 72 : GenMsgByTaskParam(rMsg);
300 72 : genTaskNum_++;
301 :
302 72 : if (rMsg->isLast != 0) {
303 37 : aivState_[aivID] = 1;
304 : }
305 72 : HCCL_INFO("CheckRcvAddrMsg, genTaskNum %u", genTaskNum_);
306 72 : rMsg->PrintMsg(GetMsgTypeString(RANK_MSG_TYPE::RANK_ADDR));
307 :
308 72 : return true;
309 : }
310 :
311 7 : bool AicpuKfcRpcServer::CheckRcvAddrMsg(HcclMsg* hcclMsg, uint32_t msgPos)
312 : {
313 7 : if (!ReadApiValidMsg(hcclMsg, &(hcclMsgArea_->commMsg.singleMsg.sendMsgs[msgPos]), false)) {
314 0 : return false;
315 : }
316 7 : AicpuKfcUtils::PrintMsg("CheckRcvAddrMsg hcclMsg", *hcclMsg);
317 7 : return true;
318 : }
319 :
320 45 : bool AicpuKfcRpcServer::ReadAddrMsg(HcclMsg* hcclMsg, uint32_t msgPos)
321 : {
322 45 : auto ctx = AicpuGetComContext();
323 45 : if (ctx == nullptr) {
324 0 : HCCL_ERROR("Get ctx is nullptr");
325 0 : return false;
326 : }
327 45 : uint32_t loopCnt = 0;
328 45 : u64 startUsec = GetCurCpuTimestamp();
329 : #ifdef CCL_LLT
330 45 : const u64 warningThreshold = static_cast<unsigned long long>(NSEC_PER_SEC);
331 45 : const u64 errorThreshold = static_cast<unsigned long long>(NSEC_PER_SEC);
332 : #else
333 : const u64 warningThreshold = static_cast<unsigned long long>(NSEC_PER_SEC) * MC2_API_MSG_TIMEOUT;
334 : const u64 errorThreshold = static_cast<unsigned long long>(NSEC_PER_SEC) * dfx::kKfcTimeOut;
335 : #endif
336 45 : u8 eventPrintTurn = 1; // 标记 Event日志的打印
337 : do {
338 31313342 : if (ctx->dfxExtendInfo.pollStatus == PollStatus::kStopAsException) {
339 1 : HCCL_ERROR("hccl aicpu exec failed, for exception.");
340 5 : return false;
341 : }
342 :
343 31313341 : KfcCommand cmd = KfcCommand::kNone;
344 31313341 : CHK_RET(AicpuHdcUtils::GetOpExecCtrlCmd(ctx->kfcControlTransferH2D, cmd));
345 31313341 : if ((cmd == KfcCommand::NsStopLaunch) && (ctx->commOpenStatus) && (!ctx->endStopLaunch)) {
346 1 : HCCL_WARNING("Stop launch due to receiving stop launch cmd.");
347 1 : AicpuUpdatComContextMumber(offsetof(AicpuComContext, isStopLaunch), true);
348 1 : AicpuUpdatComContextMumber(offsetof(AicpuComContext, endStopLaunch), true);
349 1 : return false;
350 : }
351 31313340 : if (loopCnt > 10000) { // 10000 is max loop cnt
352 3130 : loopCnt = 0;
353 : // 打印所有流的sq状态
354 3130 : HCCL_INFO(
355 : "current states %s Msg %p, msgPos %u", GetMsgTypeString(RANK_MSG_TYPE::RANK_ADDR).c_str(),
356 : &(hcclMsgArea_->commMsg.singleMsg.sendMsgs[msgPos]), msgPos);
357 : }
358 31313340 : const u64 passedTs = GetCurCpuTimestamp() - startUsec;
359 31313340 : if (passedTs > warningThreshold * eventPrintTurn) {
360 3 : HCCL_RUN_WARNING(
361 : "[AicpuKfcRpcServer][ReadAddrMsg] ReadValidMsg[%u] timeout %lus", msgPos,
362 : warningThreshold / static_cast<unsigned long long>(NSEC_PER_SEC));
363 3 : LogControl logControl(false, true);
364 3 : PrintAllHcclMsgArea();
365 3 : if (!ctx->multiServerFlag) {
366 3 : TaskOrchestrator::PrintTimeOutSqInfo(
367 : ctx, warningThreshold / static_cast<unsigned long long>(NSEC_PER_SEC));
368 : }
369 3 : eventPrintTurn *= 2; // 2 is print event log times
370 3 : if (passedTs > errorThreshold) {
371 3 : return false;
372 : }
373 3 : }
374 31313337 : loopCnt++;
375 31313337 : } while (!(ReadApiValidMsg(hcclMsg, &(hcclMsgArea_->commMsg.singleMsg.sendMsgs[msgPos]), true)));
376 :
377 : // 打印读消息的时间
378 40 : if (eventPrintTurn > 1) {
379 0 : HCCL_RUN_INFO(
380 : "[AicpuKfcRpcServer][ReadAddrMsg] Read HcclMsg[%u] cost[%llu ns]", msgPos,
381 : GetCurCpuTimestamp() - startUsec);
382 : } else {
383 40 : HCCL_INFO(
384 : "[AicpuKfcRpcServer][ReadAddrMsg] Read HcclMsg[%u] cost[%llu ns]", msgPos,
385 : GetCurCpuTimestamp() - startUsec);
386 : }
387 :
388 40 : PrintMsg(hcclMsg, msgPos);
389 40 : return true;
390 : }
391 :
392 28 : void AicpuKfcRpcServer::HcclMsg2AicAicpuOpParam(CommonHcclMsg* hcclMsg, AivAicpuOpParam* opMsg)
393 : {
394 28 : HcclApi::Mc2CcTilingInner* innerTiling = reinterpret_cast<HcclApi::Mc2CcTilingInner*>(hcclMsg->ccOpTilingData);
395 28 : AicpuComContext* ctx = AicpuGetComContext();
396 28 : if (tilingData_ == nullptr && innerTiling == nullptr) {
397 0 : HCCL_ERROR("Invalid tiling data, please check opType or other fields.");
398 0 : return;
399 : }
400 28 : opMsg->commType = hcclMsg->commType;
401 28 : opMsg->opType = hcclMsg->opType;
402 28 : opMsg->sendBuffer = hcclMsg->sendBuffer;
403 28 : opMsg->recvBuffer = hcclMsg->recvBuffer;
404 28 : opMsg->winOffset = 0U;
405 28 : opMsg->count = hcclMsg->commType == HcclCMDType::HCCL_CMD_REDUCE_SCATTER ? hcclMsg->dataCnt * ctx->rankNum :
406 : hcclMsg->dataCnt;
407 28 : opMsg->hcclDataType = hcclMsg->hcclDataType;
408 28 : opMsg->isLast = 0U;
409 28 : opMsg->sendCnt = 0x34;
410 28 : opMsg->rcvCnt = 0x12;
411 28 : opMsg->valid = hcclMsg->valid;
412 28 : opMsg->everyTurnRsp = hcclMsg->everyTurnRsp;
413 28 : opMsg->everyTurnWait = hcclMsg->everyTurnWait;
414 28 : opMsg->strideLen = static_cast<u64>(hcclMsg->strideCount);
415 :
416 28 : if (tilingData_ != nullptr) {
417 28 : opMsg->funID = tilingData_->funID;
418 28 : opMsg->totalTurnCnt = tilingData_->turnNum;
419 28 : opMsg->useBufferType = tilingData_->useBufferType;
420 : } else {
421 0 : opMsg->useBufferType = innerTiling->skipBufferWindowCopy;
422 0 : ctx->skipLocalDataCopy = innerTiling->skipLocalRankCopy;
423 : }
424 28 : if (ctx->gatherOut == 0U && opMsg->commType == HcclCMDType::HCCL_CMD_ALLGATHER) {
425 0 : ctx->gatherOut = opMsg->recvBuffer;
426 : }
427 28 : HCCL_DEBUG(
428 : "useBufferType:%u, recvBuffer[%#llx], gatherOut[%#llx], commType[%d].", opMsg->useBufferType, opMsg->recvBuffer,
429 : ctx->gatherOut, opMsg->commType);
430 : // 不需要gather out，就不需要拷贝本卡数据。需要gather out时，如果是aic负责拷贝本卡数据，reduceOp设为1即可
431 28 : if (opMsg->commType == HcclCMDType::HCCL_CMD_ALLGATHER && opMsg->opType != HCCL_REDUCE_PROD) {
432 9 : opMsg->opType = ctx->skipLocalDataCopy ? HCCL_REDUCE_PROD : HCCL_REDUCE_SUM;
433 : }
434 84 : opMsg->PrintMsg("CheckRcvAddrMsg opMsg");
435 : }
436 :
437 109 : bool AicpuKfcRpcServer::CheckAivIsEnd(uint32_t aivId) { return (aivState_[aivId] == 1); }
438 :
439 0 : bool AicpuKfcRpcServer::NeedAutoGenMsg() { return genTaskParam_ != nullptr && genTaskNum_ < tilingData_->turnNum; }
440 :
441 106 : bool AicpuKfcRpcServer::GenMsgIsLastMsg() { return (genTaskNum_ + 1 == tilingData_->turnNum); }
442 :
443 0 : uint8_t AicpuKfcRpcServer::GetWaitPolicy() { return (tilingData_->waitPolicy); }
444 :
445 37 : uint8_t AicpuKfcRpcServer::GetTaskType() const { return (tilingData_->taskType); }
446 :
447 0 : uint8_t AicpuKfcRpcServer::GetRspPolicy() { return (tilingData_->rspPolicy); }
448 :
449 0 : uint8_t AicpuKfcRpcServer::GetGenTaskNum() { return genTaskNum_; }
450 :
451 60 : TASK_PREPARE_POSITION AicpuKfcRpcServer::GetPreparePosition() const
452 : {
453 60 : return static_cast<TASK_PREPARE_POSITION>(tilingData_->preparePosition);
454 : }
455 :
456 106 : void AicpuKfcRpcServer::GenMsgByTaskParam(AivAicpuOpParam* outMsg)
457 : {
458 106 : outMsg->commType = static_cast<HcclCMDType>(tilingData_->commType);
459 106 : outMsg->opType = static_cast<HcclReduceOp>(tilingData_->reduceOp);
460 :
461 106 : switch (outMsg->commType) {
462 42 : case HcclCMDType::HCCL_CMD_ALLGATHER: {
463 42 : CalcAllgatherBuffer(outMsg);
464 :
465 : // 不需要gather out，就不需要拷贝本卡数据。需要gather out时，如果是aic负责拷贝本卡数据，reduceOp设为1即可
466 42 : if (outMsg->opType != HCCL_REDUCE_PROD) {
467 42 : outMsg->opType = tilingData_->hasCommOut ? HCCL_REDUCE_SUM : HCCL_REDUCE_PROD;
468 : }
469 42 : break;
470 : }
471 33 : case HcclCMDType::HCCL_CMD_ALLREDUCE: {
472 33 : CalcAllreduceBuffer(outMsg);
473 33 : break;
474 : }
475 28 : case HcclCMDType::HCCL_CMD_REDUCE_SCATTER: {
476 28 : CalcReduceScatterBuffer(outMsg);
477 28 : break;
478 : }
479 3 : default: {
480 3 : HCCL_ERROR("commType [%d] is not supported.", outMsg->commType);
481 3 : break;
482 : }
483 : }
484 :
485 : outMsg->count
486 106 : = genTaskNum_ < tilingData_->turnNum - tilingData_->tailNum ? tilingData_->sendCnt : tilingData_->tailSendCnt;
487 106 : outMsg->hcclDataType = static_cast<HcclDataType>(tilingData_->dataType);
488 :
489 106 : outMsg->isLast = GenMsgIsLastMsg() ? 1 : 0;
490 106 : outMsg->funID = tilingData_->funID;
491 106 : outMsg->totalTurnCnt = tilingData_->turnNum;
492 106 : outMsg->sendCnt = 0x34;
493 106 : outMsg->rcvCnt = 0x12;
494 106 : outMsg->valid = HCCL_MSG_VALID_MASK;
495 106 : outMsg->everyTurnRsp = tilingData_->rspPolicy;
496 106 : outMsg->everyTurnWait = tilingData_->waitPolicy;
497 106 : outMsg->strideLen = static_cast<u64>(tilingData_->stride);
498 106 : outMsg->useBufferType = tilingData_->useBufferType;
499 106 : }
500 :
501 105 : u64 AicpuKfcRpcServer::GetSendOff() const
502 : {
503 105 : if (tilingData_->commAlg == COMM_ALG_DOUBLE_RING || tilingData_->commAlg == COMM_ALG_SWITCH_WING) {
504 21 : return 0UL;
505 : }
506 84 : const u64 headNum = tilingData_->turnNum - tilingData_->tailNum;
507 84 : if (genTaskNum_ <= headNum) {
508 83 : return genTaskNum_ * tilingData_->sendOff;
509 : }
510 1 : return headNum * tilingData_->sendOff + (genTaskNum_ - headNum) * tilingData_->tailSendOff;
511 : }
512 :
513 105 : u64 AicpuKfcRpcServer::GetRecvOff() const
514 : {
515 105 : if (tilingData_->commAlg == COMM_ALG_DOUBLE_RING || tilingData_->commAlg == COMM_ALG_SWITCH_WING) {
516 21 : return 0UL;
517 : }
518 84 : const u64 headNum = tilingData_->turnNum - tilingData_->tailNum;
519 84 : if (genTaskNum_ <= headNum) {
520 83 : return genTaskNum_ * tilingData_->recvOff;
521 : }
522 1 : return headNum * tilingData_->recvOff + (genTaskNum_ - headNum) * tilingData_->tailRecvOff;
523 : }
524 :
525 42 : void AicpuKfcRpcServer::CalcAllgatherBuffer(AivAicpuOpParam* outMsg) const
526 : {
527 42 : const auto recvOff = GetRecvOff();
528 42 : outMsg->sendBuffer = genTaskParam_->inputA + GetSendOff();
529 42 : if (!tilingData_->useBufferType) {
530 42 : outMsg->recvBuffer = genTaskParam_->commOut + recvOff;
531 : } else {
532 0 : outMsg->recvBuffer = genTaskParam_->workSpace + tilingData_->workspaceOff + recvOff;
533 : }
534 42 : }
535 :
536 33 : void AicpuKfcRpcServer::CalcAllreduceBuffer(AivAicpuOpParam* outMsg) const
537 : {
538 33 : const auto sendOff = GetSendOff();
539 33 : const auto recvOff = GetRecvOff();
540 33 : u64 sendBuffer = 0UL;
541 33 : u64 recvBuffer = 0UL;
542 33 : if (tilingData_->commOrder == 0) { // 通信在前 或 aicpu通信展开(单allreduce)
543 21 : sendBuffer = genTaskParam_->inputA + sendOff;
544 21 : if (!tilingData_->useBufferType) {
545 15 : recvBuffer = genTaskParam_->commOut + recvOff;
546 : } else {
547 6 : recvBuffer = genTaskParam_->workSpace + tilingData_->workspaceOff + recvOff;
548 : }
549 : } else {
550 12 : sendBuffer = genTaskParam_->outputC + sendOff;
551 12 : recvBuffer = genTaskParam_->outputC + recvOff;
552 : }
553 :
554 33 : outMsg->sendBuffer = sendBuffer;
555 33 : outMsg->recvBuffer = recvBuffer;
556 33 : outMsg->winOffset = sendOff;
557 33 : }
558 :
559 28 : void AicpuKfcRpcServer::CalcReduceScatterBuffer(AivAicpuOpParam* outMsg) const
560 : {
561 28 : const auto sendOff = GetSendOff();
562 28 : const auto recvOff = GetRecvOff();
563 28 : u64 sendBuffer = 0UL;
564 28 : u64 recvBuffer = 0UL;
565 28 : if (tilingData_->commOrder == 0) { // aicpu通信展开(单reducescatter)
566 7 : sendBuffer = genTaskParam_->inputA + sendOff;
567 7 : if (!tilingData_->useBufferType) {
568 1 : recvBuffer = genTaskParam_->commOut + recvOff;
569 : } else {
570 6 : recvBuffer = genTaskParam_->workSpace + tilingData_->workspaceOff + recvOff;
571 : }
572 : } else {
573 21 : sendBuffer = genTaskParam_->workSpace + tilingData_->workspaceOff + sendOff;
574 21 : recvBuffer = genTaskParam_->outputC + recvOff;
575 : }
576 28 : outMsg->sendBuffer = sendBuffer;
577 28 : outMsg->recvBuffer = recvBuffer;
578 28 : }
579 :
580 0 : void AicpuKfcRpcServer::ClearWorkMsg() const
581 : {
582 0 : msgSndWorkArea_->sendCnt = 0;
583 0 : msgSndWorkArea_->valid = 0;
584 0 : }
585 :
586 9 : void AicpuKfcRpcServer::PrintAllHcclMsgArea()
587 : {
588 9 : const auto ctx = AicpuGetComContext();
589 9 : if (ctx == nullptr) {
590 0 : return;
591 : }
592 9 : AicpuKfcUtils::PrintAllHcclMsgArea(hcclMsgArea_, ctx->rankNum, true);
593 : }
594 :
595 40 : void AicpuKfcRpcServer::PrintMsg(HcclMsg* hcclMsg, uint32_t msgPos)
596 : {
597 40 : const auto ctx = AicpuGetComContext();
598 40 : if (ctx->debugMode == MC2_DEBUG_PRINT_MSG) {
599 2 : AicpuKfcUtils::PrintMsg("ReadAddrMsg msgPos " + std::to_string(msgPos), *hcclMsg, true);
600 2 : AicpuKfcUtils::PrintAllHcclMsgArea(hcclMsgArea_, ctx->rankNum);
601 : } else {
602 38 : AicpuKfcUtils::PrintMsg("ReadAddrMsg msgPos " + std::to_string(msgPos), *hcclMsg);
603 : }
604 :
605 40 : if (ctx->debugMode == MC2_DEBUG_PRINT_BUFF) {
606 2 : AicpuKfcUtils::PrintApiBufferByMsgPos(*hcclMsg, msgPos);
607 : }
608 40 : }
609 :
610 1 : void AicpuKfcRpcServer::PrintAllHcclMsgAreaData()
611 : {
612 65 : for (uint32_t i = 0; i < HCCL_MSG_CNT; ++i) {
613 64 : AicpuKfcUtils::PrintApiBufferByMsgPos(hcclMsgArea_->commMsg.singleMsg.sendMsgs[i], i);
614 : }
615 1 : }
|
