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_serverv2.h"
12 :
13 : #include <numeric>
14 : #include "log_control.h"
15 : #include "common/aicpu_hccl_common.h"
16 : #include "utils/hccl_aicpu_utils.h"
17 : #include "common/aicpu_kfc_utils.h"
18 : #include "framework/aicpu_kfc_prof.h"
19 :
20 : using namespace HcclApi;
21 : static constexpr uint16_t TURN_LEFT_SHIFT_BIT = 16;
22 :
23 20 : HcclResult AicpuKfcRpcServerV2::Init(const HcclMC2WorkSpace &workspaceInfo, const HcclApi::Mc2InitTilingInner *tilingData)
24 : {
25 : // 为提升效率，workspace 必须512 对齐
26 20 : u64 addr = workspaceInfo.workSpace;
27 20 : if (addr & 0x1ff) {
28 20 : addr = (addr & (~((uint64_t)0x1ff))) + 0x200;
29 : }
30 20 : Reset();
31 20 : blockNum_ = HcclAicpuUtils::GetBlockNum();
32 20 : CHK_PRT_RET(blockNum_ == 0U, HCCL_ERROR("Invalid block number."), HCCL_E_INTERNAL);
33 20 : HCCL_INFO("Align hcclmsgarea from %p to %p, block number %u, current block idx %u.",
34 : workspaceInfo.workSpace, addr, blockNum_, HcclAicpuUtils::GetBlockIdx());
35 20 : if (tilingData != nullptr && tilingData->queueNum > 0U) {
36 2 : totalQueueNum_ = tilingData->commBlockNum * tilingData->queueNum;
37 2 : CHK_PRT_RET(totalQueueNum_ > LOCAL_STREAM_MAX_NUM || blockNum_ > std::min(MAX_AICPU_NUM_BLOCKS, totalQueueNum_),
38 : HCCL_ERROR("Invalid para, comm block %u, aicpu block %u, queue number %u.",
39 : tilingData->commBlockNum, blockNum_, tilingData->queueNum), HCCL_E_INTERNAL);
40 : } else {
41 18 : totalQueueNum_ = 0U;
42 : }
43 20 : hcclMsgArea_ = reinterpret_cast<HcclMsgArea *>(addr);
44 20 : turnNumAddr_ = addr + sizeof(HcclMsgArea);
45 20 : if (turnNumAddr_ + sizeof(u32) * TILING_TURN_MAX * HCCL_MAX_RANK_NUM_V2 >
46 20 : workspaceInfo.workSpace + workspaceInfo.workSpaceSize) {
47 0 : HCCL_ERROR("Turn number addr %#llx, space for turn number is %lu, the space after workspace %#llx will "
48 : "be overwritten.", turnNumAddr_, sizeof(u32) * TILING_TURN_MAX * HCCL_MAX_RANK_NUM_V2,
49 : workspaceInfo.workSpace + workspaceInfo.workSpaceSize);
50 0 : return HCCL_E_INTERNAL;
51 : }
52 20 : uint32_t *turnNums = reinterpret_cast<uint32_t *>(turnNumAddr_);
53 20 : std::iota(&turnNums[0], &turnNums[TILING_TURN_MAX * HCCL_MAX_RANK_NUM_V2], 0);
54 20 : tilingBaseAddr_ = reinterpret_cast<u64>(tilingData);
55 20 : return HCCL_SUCCESS;
56 : }
57 :
58 50 : void AicpuKfcRpcServerV2::GetLocalQueueRange(u32 &start, u32 &end)
59 : {
60 50 : if (blockNum_ == 0U || totalQueueNum_ == 0U) {
61 40 : start = end = 0U;
62 40 : return;
63 : }
64 10 : const u32 base = totalQueueNum_ / blockNum_;
65 10 : const u32 remainder = totalQueueNum_ % blockNum_;
66 10 : const u32 blockIdx = HcclAicpuUtils::GetBlockIdx();
67 10 : if (blockIdx < remainder) {
68 0 : start = blockIdx * base + blockIdx;
69 0 : end = start + base;
70 : } else {
71 10 : start = blockIdx * base + remainder;
72 10 : end = start + base - 1U;
73 : }
74 : }
75 :
76 20 : void AicpuKfcRpcServerV2::Reset()
77 : {
78 20 : (void)memset_s(msgPos_, sizeof(msgPos_), 0, sizeof(msgPos_));
79 20 : msgPosForKernel_ = 0;
80 1280 : for (int8_t i = 0; i < HCCL_MAX_HANDLE_ID; i++) {
81 1260 : handleIdMsgPosition_[i] = -1;
82 : }
83 20 : (void)memset_s(isFinalize_, sizeof(isFinalize_), 0, sizeof(isFinalize_));
84 20 : (void)memset_s(barrierFlags_, sizeof(barrierFlags_), 0, sizeof(barrierFlags_));
85 20 : (void)memset_s(barrierFinishCnt_, sizeof(barrierFinishCnt_), 0, sizeof(barrierFinishCnt_));
86 20 : const u64 ts = GetCurCpuTimestamp();
87 980 : for (u32 i = 0U; i < MAX_QUE_NUM; ++i) {
88 960 : prepareTime_[i] = ts;
89 : }
90 20 : eventPrintTurn_ = 1U;
91 20 : }
92 :
93 9 : void AicpuKfcRpcServerV2::SetMsgHandlePos(uint32_t msgPos, HcclHandle handleId) {
94 9 : if (handleId >= HCCL_MAX_HANDLE_ID || handleId < 0) {
95 0 : return;
96 : }
97 9 : handleIdMsgPosition_[handleId] = msgPos;
98 : }
99 :
100 7 : int32_t AicpuKfcRpcServerV2::GetMsgHandlePos(HcclHandle handleId) {
101 7 : if (handleId >= HCCL_MAX_HANDLE_ID || handleId < 0) {
102 0 : HCCL_ERROR("[GetMsgHandlePos] invalid handleId %d", handleId);
103 0 : return -1;
104 : }
105 7 : if (handleIdMsgPosition_[handleId] < 0) {
106 5 : HCCL_WARNING("[GetMsgHandlePos] invalid handleIdMsgPosition %d", handleIdMsgPosition_[handleId]);
107 5 : return -1;
108 : }
109 2 : return handleIdMsgPosition_[handleId];
110 : }
111 :
112 1 : bool AicpuKfcRpcServerV2::IsPrintLog() const {
113 1 : return isPrintLog_;
114 : }
115 :
116 61 : bool AicpuKfcRpcServerV2::GetIsFinalize(u32 queueId) {
117 61 : if (queueId < MAX_QUE_NUM) {
118 28 : return isFinalize_[queueId];
119 : }
120 33 : u32 start = 0U;
121 33 : u32 end = 0U;
122 33 : GetLocalQueueRange(start, end);
123 43 : for (u32 i = start; i <= end; ++i) {
124 34 : if (!isFinalize_[i]) {
125 24 : return false;
126 : }
127 : }
128 9 : return true;
129 : }
130 :
131 10 : void AicpuKfcRpcServerV2::SetIsFinalize(u32 queueId, bool finalize) {
132 10 : isFinalize_[queueId] = finalize;
133 10 : }
134 :
135 7 : HcclMsgExt* AicpuKfcRpcServerV2::GetHcclMsgExtPtr() {
136 7 : return msgExt_.get();
137 : }
138 :
139 0 : HcclMsgArea* AicpuKfcRpcServerV2::GetHcclMsgArea(void) {
140 0 : return hcclMsgArea_;
141 : }
142 :
143 29 : HcclMsg (*AicpuKfcRpcServerV2::GetMsgWorkSpace())[HCCL_MSG_CNT] {
144 29 : if (totalQueueNum_ == 0U) {
145 24 : return &(hcclMsgArea_->commMsg.singleMsg.sendMsgs);
146 : } else {
147 5 : return hcclMsgArea_->commMsg.multiMsg.sendMsgs;
148 : }
149 : }
150 :
151 7 : uint64_t AicpuKfcRpcServerV2::GetFinishAddr(int32_t idx) const {
152 7 : if (idx >= static_cast<int32_t>(HCCL_MSG_CNT) || hcclMsgArea_ == nullptr) {
153 3 : HCCL_ERROR("idx %d exceed max or msg area is not initialized.", idx);
154 3 : return 0;
155 : }
156 4 : return reinterpret_cast<uint64_t>(&(hcclMsgArea_->commMsg.singleMsg.finishedTurnCnt[idx].cnt));
157 : }
158 :
159 14 : uint64_t AicpuKfcRpcServerV2::GetCommitareaAddr(int32_t idx) const {
160 14 : if (idx >= static_cast<int32_t>(HCCL_MSG_CNT) || hcclMsgArea_ == nullptr) {
161 3 : HCCL_ERROR("idx %d exceed max or hcclMsgArea_ is not initialized.", idx);
162 3 : return 0;
163 : }
164 11 : return reinterpret_cast<uint64_t>(&(hcclMsgArea_->commMsg.singleMsg.commitTurnCnt[idx].cnt));
165 : }
166 :
167 12 : HcclResult AicpuKfcRpcServerV2::AddFlipTask(HcclDispatcher dispatcherPtr, hccl::Stream *stream)
168 : {
169 12 : if (!dfx::ProfilingManager::GetProfL0State()) {
170 12 : return HCCL_SUCCESS;
171 : }
172 0 : hccl::HcclSqeContext *sqeCtx = stream->GetSqeContextPtr();
173 0 : CHK_PTR_NULL(sqeCtx);
174 0 : hccl::SqeRingBuffer &buff = sqeCtx->buffer;
175 : // nextTaskId=0的时候下发PlaceHolder
176 0 : if (UNLIKELY(buff.tailSqeTaskId == 0 && buff.filpNum != 0)) {
177 0 : CHK_RET(AddRetryPreamble(dispatcherPtr, *stream));
178 : }
179 :
180 0 : return HCCL_SUCCESS;
181 : }
182 :
183 9 : HcclResult AicpuKfcRpcServerV2::AddCcoreWait(HcclDispatcher dispatcherPtr, u64 waitAddr, uint32_t turnNum,
184 : hccl::Stream *stream, bool isLast) // client commit wait
185 : {
186 9 : uint8_t *sqeBuffer = nullptr;
187 9 : uint8_t *sqeTypeAddr = nullptr;
188 9 : uint8_t *sqeDfxInfoAddr = nullptr;
189 9 : uint16_t taskId = 0U;
190 :
191 9 : CHK_RET(AddFlipTask(dispatcherPtr, stream));
192 9 : CHK_RET(stream->GetNextSqeBufferAddr(sqeBuffer, sqeTypeAddr, sqeDfxInfoAddr, taskId));
193 9 : const HcclComStreamInfo &streamInfo = stream->GetHcclStreamInfo();
194 9 : if (AicpuKfcProf::IsDebugModeEquals(MC2_DEBUG_COMMIT_TIMEOUT)) {
195 0 : uint32_t *turnNums = reinterpret_cast<uint32_t *>(turnNumAddr_);
196 0 : turnNums[turnNum] = 0xFF;
197 : }
198 9 : AddOneWaitStartSqe(streamInfo.actualStreamId, taskId, waitAddr, turnNumAddr_ + turnNum * sizeof(u32),
199 : isLast, reinterpret_cast<rtStarsCcoreWaitStartSqe_t *>(sqeBuffer), sqeTypeAddr);
200 9 : hccl::HcclSqeContext* sqeCtx = stream->GetSqeContextPtr();
201 9 : if (sqeCtx == nullptr) {
202 0 : HCCL_ERROR("AddCcoreWait sqeCtx is nullptr");
203 0 : return HCCL_E_INTERNAL;
204 : }
205 9 : sqeCtx->buffer.addInfo[taskId % hccl::HCCL_SQE_MAX_CNT] =
206 9 : ((turnNum << TURN_LEFT_SHIFT_BIT) + static_cast<uint32_t>(isLast));
207 9 : return HCCL_SUCCESS;
208 : }
209 :
210 3 : HcclResult AicpuKfcRpcServerV2::AddCcoreNotify(HcclDispatcher dispatcherPtr, u64 recordAddr, uint32_t turnNum,
211 : hccl::Stream *stream) // client finish notify
212 : {
213 3 : uint8_t *sqeBuffer = nullptr;
214 3 : uint8_t *sqeTypeAddr = nullptr;
215 3 : uint8_t *sqeDfxInfoAddr = nullptr;
216 3 : uint16_t taskId = 0U;
217 :
218 3 : CHK_RET(AddFlipTask(dispatcherPtr, stream));
219 3 : CHK_RET(stream->GetNextSqeBufferAddr(sqeBuffer, sqeTypeAddr, sqeDfxInfoAddr, taskId));
220 3 : const HcclComStreamInfo &streamInfo = stream->GetHcclStreamInfo();
221 3 : if (AicpuKfcProf::IsDebugModeEquals(MC2_DEBUG_AICORE_WAIT_TIMEOUT)) {
222 0 : uint32_t *turnNums = reinterpret_cast<uint32_t *>(turnNumAddr_);
223 0 : turnNums[turnNum] = 0;
224 : }
225 3 : AddOneWriteValueStartSqe(streamInfo.actualStreamId, taskId, recordAddr,
226 3 : turnNumAddr_ + turnNum * sizeof(u32), reinterpret_cast<rtStarsCcoreWriteValueSqe_t *>(sqeBuffer),
227 : sqeTypeAddr);
228 3 : hccl::HcclSqeContext* sqeCtx = stream->GetSqeContextPtr();
229 3 : if (sqeCtx == nullptr) {
230 3 : HCCL_ERROR("AddCcoreNotify sqeCtx is nullptr");
231 3 : return HCCL_E_INTERNAL;
232 : }
233 0 : sqeCtx->buffer.addInfo[taskId % hccl::HCCL_SQE_MAX_CNT] = turnNum;
234 0 : return HCCL_SUCCESS;
235 : }
236 :
237 3 : uint64_t AicpuKfcRpcServerV2::GetFinishAddrByHandleId(HcclHandle handleId)
238 : {
239 3 : int32_t msgPos = GetMsgHandlePos(handleId);
240 3 : if (msgPos < 0) {
241 2 : return 0;
242 : }
243 1 : return GetFinishAddr(msgPos);
244 : }
245 :
246 12 : void AicpuKfcRpcServerV2::SetMsgRepeatCnt(u8 repeatCnt)
247 : {
248 12 : repeatCnt_[msgPos_[0U]] = (totalStep_ == 0U ? repeatCnt : repeatCnt * totalStep_);
249 12 : }
250 :
251 4 : int32_t AicpuKfcRpcServerV2::GetMsgRepeatCnt(HcclHandle handleId)
252 : {
253 4 : int32_t msgPos = GetMsgHandlePos(handleId);
254 4 : if (msgPos < 0) {
255 3 : return -1;
256 : }
257 1 : return repeatCnt_[msgPos];
258 : }
259 :
260 12 : HcclResult AicpuKfcRpcServerV2::ProcessExpectPrepareMsg(uint8_t seqNum, uint8_t expectId)
261 : {
262 : // 当前无翻转场景，只需考虑单个通信域提前Finializa场景
263 12 : if (seqNum == 0 && expectId > 0) {
264 1 : return HCCL_SUCCESS;
265 : }
266 11 : if (seqNum < expectId) {
267 1 : HCCL_ERROR("curMsg seqNum %d is smaller than expect %d ignore.", seqNum, expectId);
268 1 : return HCCL_E_INTERNAL;
269 : }
270 10 : if (totalQueueNum_ == 0U && seqNum > expectId) {
271 1 : HCCL_INFO("curMsg seqNum %d is bigger than expect %d ignore.", seqNum, expectId);
272 1 : return HCCL_E_UNAVAIL;
273 : }
274 9 : return HCCL_SUCCESS;
275 : }
276 :
277 15 : void AicpuKfcRpcServerV2::SetNeedRetryFlag(bool needRetryFlag)
278 : {
279 15 : needReProcess_ = needRetryFlag;
280 15 : }
281 :
282 4 : bool AicpuKfcRpcServerV2::ReadValidMsgExtArea(int32_t idx, u32 rankSize)
283 : {
284 : #ifdef __aarch64__
285 : __asm__ __volatile__("dsb ld" : : : "memory");
286 : #endif
287 : #ifdef __amd64__
288 4 : __asm__ __volatile__("" : : : "memory");
289 : #endif
290 4 : auto &extMsgList = hcclMsgArea_->commMsg.singleMsg.paramExtMsgList[idx];
291 4 : if (hcclMsgArea_ == nullptr || extMsgList.valid != static_cast<u64>(HCCL_MSG_VALID_MASK)) {
292 2 : return false;
293 : }
294 2 : uint64_t msgExtXorCheck = AicpuKfcUtils::GenXor(&extMsgList, rankSize);
295 : static uint32_t msgExtXorCheckTurn = 0;
296 2 : if (UNLIKELY(msgExtXorCheck != extMsgList.xorCheck)) {
297 1 : if (msgExtXorCheckTurn++ % MC2_API_XORCHECK_PRINT_NUM == 0) {
298 1 : HCCL_RUN_INFO("Extend data is modified! modified_xor:%llu, origin_xor:%llu.",
299 : msgExtXorCheck, extMsgList.xorCheck);
300 : }
301 1 : return false;
302 : }
303 1 : HCCL_INFO("hcclMsgArea xorCheck[%llu]", extMsgList.xorCheck);
304 1 : const size_t copySize = sizeof(uint64_t) * rankSize;
305 1 : (void)memcpy_s(msgExt_->sendCounts, copySize, extMsgList.sendCounts, copySize);
306 1 : (void)memcpy_s(msgExt_->sendOffset, copySize, extMsgList.sendOffset, copySize);
307 1 : (void)memcpy_s(msgExt_->recvCounts, copySize, extMsgList.recvCounts, copySize);
308 1 : (void)memcpy_s(msgExt_->recvOffset, copySize, extMsgList.recvOffset, copySize);
309 1 : (void)memcpy_s(msgExt_->reserved, sizeof(HcclMsgExt) - offsetof(HcclMsgExt, reserved),
310 1 : extMsgList.reserved, sizeof(HcclMsgExt) - offsetof(HcclMsgExt, reserved));
311 :
312 : #ifdef __aarch64__
313 : __asm__ __volatile__("dsb ld" : : : "memory");
314 : #endif
315 : #ifdef __amd64__
316 1 : __asm__ __volatile__("" : : : "memory");
317 : #endif
318 1 : extMsgList.valid = static_cast<u64>(~HCCL_MSG_VALID_MASK);
319 : #ifdef __aarch64__
320 : __asm__ __volatile__("dsb st" : : : "memory");
321 : #endif
322 1 : HCCL_INFO("reset paramExtMsgList valid value %lu", extMsgList.valid);
323 1 : return true;
324 : }
325 :
326 15 : bool AicpuKfcRpcServerV2::IsExceedLimit(HcclCMDType commType, u32 rankSize)
327 : {
328 15 : if (rankSize > HCCL_MAX_RANK_NUM_V2 && (commType == HcclCMDType::HCCL_CMD_ALLTOALLV || commType == HcclCMDType::HCCL_CMD_ALLTOALL)) {
329 0 : HCCL_ERROR("The number[%u] of ranks exceeds the 256p limit supported by the ALLTOALL/ALLTOALLV algorithm.",
330 : rankSize);
331 0 : return true;
332 : }
333 15 : return false;
334 : }
335 :
336 18 : bool AicpuKfcRpcServerV2::ReadValidMsg(HcclMsg *rMsg, HcclMsg *msg, bool needReProcess, uint32_t msgPos, u32 rankSize)
337 : {
338 : #ifdef __aarch64__
339 : __asm__ __volatile__("dsb ld" : : : "memory");
340 : #endif
341 : #ifdef __amd64__
342 18 : __asm__ __volatile__("" : : : "memory");
343 : #endif
344 : // 重处理消息
345 18 : if (needReProcess) {
346 0 : *rMsg = *msg;
347 0 : return true;
348 : }
349 18 : if (msg->addMsg.v0Msg.valid != HCCL_MSG_VALID_MASK) {
350 2 : return false;
351 : }
352 16 : memcpy_s(rMsg, sizeof(HcclMsg), msg, sizeof(HcclMsg));
353 16 : uint32_t msgXorCheck = AicpuKfcUtils::GenXor(rMsg);
354 : static uint32_t msgXorCheckTurn = 0;
355 16 : if (UNLIKELY(msgXorCheck != rMsg->addMsg.v0Msg.xorCheck)) {
356 2 : if (msgXorCheckTurn++ % MC2_API_XORCHECK_PRINT_NUM == 0) {
357 2 : AicpuKfcUtils::PrintMsg("Rcv src msg", *msg, true);
358 1 : AicpuKfcUtils::PrintMsg("Rcv dst msg", *rMsg, true);
359 1 : HCCL_RUN_INFO("data is modified! modified_xor:%u, origin_xor:%u.", msgXorCheck,
360 : rMsg->addMsg.v0Msg.xorCheck);
361 : }
362 2 : return false;
363 : }
364 14 : if (UNLIKELY(IsExceedLimit(static_cast<HcclCMDType>(rMsg->commType.prepareType), rankSize))) {
365 0 : return false;
366 : }
367 14 : if (UNLIKELY(static_cast< HcclCMDType>(rMsg->commType.prepareType) == HCCL_CMD_ALLTOALLV &&
368 : !ReadValidMsgExtArea(msgPos, rankSize))) {
369 1 : return false;
370 : }
371 13 : msg->addMsg.v0Msg.valid = ~HCCL_MSG_VALID_MASK;
372 13 : if (UNLIKELY(AicpuKfcProf::IsDebugModeEquals(MC2_DEBUG_PREPARE_TIMEOUT) &&
373 : (rMsg->commType.msgType != ControlMsgType::HCCL_CMD_FINALIZE))) {
374 0 : return false;
375 : }
376 13 : if (UNLIKELY(AicpuKfcProf::IsDebugModeEquals(MC2_DEBUG_FINALIZE_TIMEOUT) &&
377 : (rMsg->commType.msgType == ControlMsgType::HCCL_CMD_FINALIZE))) {
378 0 : return false;
379 : }
380 13 : HCCL_INFO("reset valid value 0x%x", msg->addMsg.v0Msg.valid);
381 13 : return true;
382 : }
383 :
384 17 : bool AicpuKfcRpcServerV2::ReadAddrMsg(HcclMsg *hcclMsg, HcclMsg *msgList, u32 queueIdx, u32 msgPos, u32 rankSize)
385 : {
386 17 : bool ret = ReadValidMsg(hcclMsg, &(msgList[msgPos]), needReProcess_, msgPos, rankSize);
387 17 : isPrintLog_ = false;
388 17 : if (LIKELY(ret)) {
389 13 : HCCL_DEBUG("read valid msg msgPos %u commType %u", msgPos, static_cast<uint32_t>(hcclMsg->commType.msgType));
390 13 : PrintMsg(hcclMsg, msgPos, rankSize);
391 : // Prepare 成功，打印耗时
392 13 : u64 prepareTime = GetCurCpuTimestamp();
393 13 : if (eventPrintTurn_ > 1) {
394 0 : HCCL_RUN_INFO("[AicpuKfcRpcServerV2][ReadAddrMsg] Read HcclMsg[%u] cost %llu",
395 : msgPos, prepareTime - prepareTime_[queueIdx]);
396 : } else {
397 13 : HCCL_INFO("[AicpuKfcRpcServerV2][ReadAddrMsg] Read HcclMsg[%u] cost %llu",
398 : msgPos, prepareTime - prepareTime_[queueIdx]);
399 : }
400 13 : prepareTime_[queueIdx] = prepareTime;
401 13 : eventPrintTurn_ = 1;
402 4 : } else if (GetCurCpuTimestamp() - prepareTime_[queueIdx] >
403 4 : static_cast<unsigned long long>(NSEC_PER_SEC) * MC2_API_MSG_TIMEOUT * eventPrintTurn_) {
404 : // Prepare 等待 20s
405 0 : HCCL_RUN_WARNING("[AicpuKfcRpcServerV2][ReadAddrMsg] ReadValidMsg[%u] timeout %lus", msgPos,
406 : MC2_API_MSG_TIMEOUT * eventPrintTurn_);
407 0 : eventPrintTurn_ *= 2; // 2 is print event log times
408 0 : LogControl logControl(false, true);
409 0 : PrintAllHcclMsgArea(rankSize);
410 0 : isPrintLog_ = true;
411 0 : }
412 17 : return ret;
413 : }
414 :
415 : // reset消息区msgPos的commitTurnId
416 8 : HcclResult AicpuKfcRpcServerV2::ResetCommitTaskAdd(HcclDispatcher dispatcherPtr, hccl::Stream *stream)
417 : {
418 : // reset函数复用 AddCcoreWait,turnNum保证条件算子恒成立
419 16 : for (uint32_t i = 0; i < static_cast<uint32_t>(msgPos_[0U]); i++) {
420 8 : uint64_t waitAddr = GetCommitareaAddr(i);
421 8 : CHK_RET(AddCcoreWait(dispatcherPtr, waitAddr, 0, stream, true));
422 : }
423 8 : return HCCL_SUCCESS;
424 : }
425 :
426 9 : void AicpuKfcRpcServerV2::WriteFinishWhenAllFinalize()
427 : {
428 9 : if (hcclMsgArea_ == nullptr || totalQueueNum_ != 0U) {
429 1 : return;
430 : }
431 8 : uint32_t msgPos = GetMsgPos();
432 8 : hcclMsgArea_->commMsg.singleMsg.finishedTurnCnt[msgPos].cnt = FINALIZE_FINISH_CNT; // 用于校验的非法值
433 8 : HCCL_INFO("Post finishedTurnCnt[%u].cnt = %lu.",
434 : msgPos, hcclMsgArea_->commMsg.singleMsg.finishedTurnCnt[msgPos].cnt);
435 : #ifdef __aarch64__
436 : __asm__ __volatile__("dsb st" : : : "memory");
437 : #endif
438 : }
439 :
440 0 : void AicpuKfcRpcServerV2::WriteRestartFlag()
441 : {
442 0 : if (totalQueueNum_ == 0U) {
443 0 : for (uint32_t i = 0; i < HCCL_MSG_CNT; i++) {
444 0 : hcclMsgArea_->commMsg.singleMsg.sendMsgs[i].addMsg.v0Msg.valid = ~HCCL_MSG_VALID_MASK;
445 0 : hcclMsgArea_->commMsg.singleMsg.commitTurnCnt[i].cnt = 0;
446 0 : hcclMsgArea_->commMsg.singleMsg.finishedTurnCnt[i].cnt = 0;
447 : }
448 : } else {
449 0 : for (uint32_t i = 0; i < MAX_QUE_NUM; i++) {
450 0 : for (uint32_t j = 0; j < HCCL_MSG_CNT; j++) {
451 0 : hcclMsgArea_->commMsg.multiMsg.sendMsgs[i][j].addMsg.v1Msg.valid = ~HCCL_MSG_VALID_MASK;
452 : }
453 : }
454 : }
455 0 : hcclMsgArea_->controlMsg.restart = 1;
456 : #ifdef __aarch64__
457 : __asm__ __volatile__("dsb st" : : : "memory");
458 : #endif
459 0 : }
460 :
461 2 : void AicpuKfcRpcServerV2::PrintAllHcclMsgArea(u32 rankSize)
462 : {
463 2 : if (totalQueueNum_ == 0U) {
464 1 : AicpuKfcUtils::PrintAllHcclMsgArea(hcclMsgArea_, rankSize, true);
465 1 : } else if (HcclAicpuUtils::GetBlockIdx() == 0U) {
466 1 : AicpuKfcUtils::PrintAllHcclMsgAreaForMulti(hcclMsgArea_, true);
467 : }
468 2 : }
469 :
470 13 : void AicpuKfcRpcServerV2::PrintMsg(HcclMsg *hcclMsg, uint32_t msgPos, u32 rankSize)
471 : {
472 13 : if (AicpuKfcProf::IsDebugModeEquals(MC2_DEBUG_PRINT_MSG)) {
473 0 : AicpuKfcUtils::PrintMsg("ReadAddrMsg msgPos " + std::to_string(msgPos), *hcclMsg, true);
474 0 : if (totalQueueNum_ == 0U) {
475 0 : AicpuKfcUtils::PrintAllHcclMsgArea(hcclMsgArea_, rankSize);
476 : } else {
477 0 : AicpuKfcUtils::PrintAllHcclMsgAreaForMulti(hcclMsgArea_);
478 : }
479 : } else {
480 13 : AicpuKfcUtils::PrintMsg("ReadAddrMsg msgPos " + std::to_string(msgPos), *hcclMsg);
481 : }
482 13 : if (AicpuKfcProf::IsDebugModeEquals(MC2_DEBUG_PRINT_BUFF)) {
483 13 : AicpuKfcUtils::PrintApiBufferByMsgPos(*hcclMsg, msgPos);
484 : }
485 13 : }
486 :
487 10 : void AicpuKfcRpcServerV2::PrintAllHcclMsgAreaData()
488 : {
489 10 : if (totalQueueNum_ != 0U) {
490 2 : return;
491 : }
492 520 : for (uint32_t i = 0; i < HCCL_MSG_CNT; ++i) {
493 512 : AicpuKfcUtils::PrintApiBufferByMsgPos(hcclMsgArea_->commMsg.singleMsg.sendMsgs[i], i);
494 : }
495 : }
496 :
497 1 : void AicpuKfcRpcServerV2::DumpBarrierInfo(u32 groupIdx, u32 sqId, u32 devId)
498 : {
499 1 : const auto &barrierInfos = barrierFlags_[groupIdx];
500 3 : for (u32 i = 0U; i < totalQueueNum_; ++i) {
501 2 : const BarrierStatus status = barrierInfos[i].status;
502 2 : HCCL_ERROR("Queue:%u, msg pos:%u, status:%u.", i, GetMsgPos(i), static_cast<u32>(status));
503 2 : if (status == BarrierStatus::SELF_BARRIER) {
504 : u32 sqHead, sqTail;
505 0 : (void)QuerySqStatusByType(devId, sqId, DRV_SQCQ_PROP_SQ_HEAD, sqHead);
506 0 : (void)QuerySqStatusByType(devId, sqId, DRV_SQCQ_PROP_SQ_TAIL, sqTail);
507 0 : HCCL_ERROR("Queue:%u, sq head:%u, tail:%u.", i, sqHead, sqTail);
508 : }
509 : }
510 1 : }
|
