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

Generated by: LCOV version 2.0-1