LCOV - code coverage report
Current view: top level - legacy/ascend910/framework/device/aicpu_kfc/framework - aicpu_kfc_rpc_serverv2.cc (source / functions) Coverage Total Hit
Test: coverage.info Lines: 81.2 % 292 237
Test Date: 2026-07-28 12:11:00 Functions: 93.8 % 32 30

            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 : }
        

Generated by: LCOV version 2.0-1