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