LCOV - code coverage report
Current view: top level - base_comm/resources/endpoint_pairs/channels/host - host_cpu_urma_channel.cc (source / functions) Coverage Total Hit
Test: coverage.info Lines: 58.1 % 241 140
Test Date: 2026-08-18 17:47:01 Functions: 62.5 % 24 15

            Line data    Source code
       1              : /**
       2              :  * Copyright (c) 2026 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 "host_cpu_urma_channel.h"
      12              : #include "endpoint.h"
      13              : #include "orion_adpt_utils.h"
      14              : #include "hcomm_adapter_urma.h"
      15              : 
      16              : // Orion
      17              : #include "topo_common_types.h"
      18              : #include "virtual_topo.h"
      19              : #include "env_config/env_config_v2.h"
      20              : 
      21              : namespace hcomm {
      22              : constexpr uint16_t DEFAULT_LISTENING_PORT = 60001;
      23              : 
      24           17 : HostCpuUrmaChannel::HostCpuUrmaChannel(EndpointHandle endpointHandle, const HcommChannelDesc& channelDesc)
      25           17 :     : endpointHandle_(endpointHandle),
      26           17 :       channelDesc_(channelDesc)
      27           17 : {}
      28              : 
      29           34 : HostCpuUrmaChannel::~HostCpuUrmaChannel()
      30              : {
      31           17 :     if (channelDesc_.socket == nullptr && socket_ != nullptr) {
      32            1 :         SocketMgr::GetInstance(devicePhyId_).PutSocket(socketConfig_, socket_);
      33            1 :         socket_ = nullptr;
      34              :     }
      35           34 : }
      36              : 
      37           10 : HcclResult HostCpuUrmaChannel::ParseInputParam()
      38              : {
      39              :     // 1. 从 endpointHandle_,获得 localEp_ 和 rdmaHandle_
      40           10 :     Endpoint* localEpPtr = reinterpret_cast<Endpoint*>(endpointHandle_);
      41           10 :     CHK_PTR_NULL(localEpPtr);
      42            9 :     localEp_ = localEpPtr->GetEndpointDesc();
      43            9 :     rdmaHandle_ = localEpPtr->GetRdmaHandle();
      44              : 
      45            9 :     HCCL_INFO("[HostCpuUrmaChannel][%s] localProtocol[%d]", __func__, localEp_.protocol);
      46              : 
      47              :     // 2. 从 channelDesc_,获得 remoteEp_, socket_ 和 notifyNum
      48            9 :     remoteEp_ = channelDesc_.remoteEndpoint;
      49            9 :     socket_ = reinterpret_cast<Hccl::Socket*>(channelDesc_.socket);
      50            9 :     commonRes_.bufferVec.clear();
      51              : 
      52            9 :     if (channelDesc_.exchangeAllMems) {
      53              :         // 3. Get memHandles from endpoint
      54            1 :         HCCL_INFO("[HostCpuUrmaChannel][%s] exchangeAllMems == True. Get memHandles from endpoint.", __func__);
      55            1 :         std::shared_ptr<Hccl::LocalUbRmaBuffer>* memHandles = nullptr;
      56            1 :         uint32_t memHandleNum = 0;
      57            1 :         CHK_RET(static_cast<HcclResult>(
      58              :             HcommMemGetAllMemHandles(endpointHandle_, reinterpret_cast<void**>(&memHandles), &memHandleNum)));
      59            1 :         HCCL_INFO("[HostCpuUrmaChannel][%s] Got memHandleNum[%u].", __func__, memHandleNum);
      60            2 :         for (uint32_t i = 0; i < memHandleNum; ++i) {
      61            1 :             std::shared_ptr<Hccl::LocalUbRmaBuffer>& localUbRmaBuffer = memHandles[i];
      62            1 :             CHK_SMART_PTR_NULL(localUbRmaBuffer);
      63            1 :             Hccl::Buffer* buf = localUbRmaBuffer->GetBuf();
      64            1 :             CHK_PTR_NULL(buf);
      65            1 :             HCCL_INFO(
      66              :                 "[HostCpuUrmaChannel][%s] Got memHandle No.%u: addr[0x%llx], size[0x%llx], memType[%d], memInfo[%s].",
      67              :                 __func__, i, static_cast<unsigned long long>(localUbRmaBuffer->GetAddr()),
      68              :                 static_cast<unsigned long long>(localUbRmaBuffer->GetSize()), static_cast<int>(buf->GetMemType()),
      69              :                 buf->GetMemInfo().c_str());
      70            1 :             commonRes_.bufferVec.push_back(localUbRmaBuffer.get());
      71              :         }
      72              :     } else {
      73              :         // 3. 从 channelDesc 的 memHandle,获得 bufs_
      74            8 :         HCCL_WARNING("[HostCpuUrmaChannel][%s] exchangeAllMems is false.", __func__);
      75              :     }
      76              : 
      77            9 :     return HCCL_SUCCESS;
      78              : }
      79              : 
      80            1 : HcclResult HostCpuUrmaChannel::StartListen()
      81              : {
      82            1 :     uint16_t port = channelDesc_.port;
      83            1 :     HCCL_INFO(
      84              :         "[HostCpuUrmaChannel::%s] Start. EndpointHandle[0x%llx], port[%u]", __func__,
      85              :         reinterpret_cast<uint64_t>(endpointHandle_), port);
      86            1 :     if (port == 0) {
      87            0 :         port = DEFAULT_LISTENING_PORT;
      88            0 :         HCCL_INFO("[HostCpuUrmaChannel::%s] channelDesc port is 0, use default port [%u]", __func__, port);
      89              :     }
      90            1 :     CHK_RET(static_cast<HcclResult>(HcommEndpointStartListen(endpointHandle_, port, nullptr)));
      91            0 :     HCCL_INFO("[HostCpuUrmaChannel::%s] SUCCESS. port[%u].", __func__, port);
      92            0 :     return HCCL_SUCCESS;
      93              : }
      94              : 
      95            8 : HcclResult HostCpuUrmaChannel::BuildSocket()
      96              : {
      97            8 :     if (socket_ != nullptr) {
      98            7 :         return HCCL_SUCCESS;
      99              :     }
     100            1 :     HCCL_INFO("[HostCpuUrmaChannel::%s] socket ptr is NULL, rebuild Socket", __func__);
     101              : 
     102            1 :     Hccl::LinkData linkData = BuildDefaultLinkData();
     103            1 :     CHK_RET(EndpointDescPairToLinkData(localEp_, remoteEp_, linkData));
     104            1 :     HCCL_INFO("[HostCpuUrmaChannel::%s] built linkData: %s", __func__, linkData.Describe().c_str());
     105            1 :     uint16_t port = channelDesc_.port;
     106            1 :     if (port == 0) {
     107            0 :         port = DEFAULT_LISTENING_PORT;
     108            0 :         HCCL_INFO("[HostCpuUrmaChannel::%s] channelDesc port is 0, use default port [%u]", __func__, port);
     109              :     }
     110              : 
     111              :     std::string socketTag
     112            3 :         = (channelDesc_.channelName != nullptr) ? std::string(channelDesc_.channelName) : "AUTOMATIC_SOCKET_TAG";
     113              :     Hccl::SocketConfig socketConfig
     114            1 :         = (channelDesc_.role != HCOMM_SOCKET_ROLE_RESERVED) ?
     115            1 :               Hccl::SocketConfig(linkData, port, socketTag, channelDesc_.role == HCOMM_SOCKET_ROLE_SERVER) :
     116            1 :               Hccl::SocketConfig(linkData, socketTag, true);
     117            1 :     CHK_RET(SocketMgr::GetInstance(devicePhyId_).GetSocket(socketConfig, socket_));
     118              : 
     119            1 :     HCCL_INFO("[HostCpuUrmaChannel::%s] SUCCESS. port[%u].", __func__, port);
     120            1 :     return HCCL_SUCCESS;
     121            1 : }
     122              : 
     123            8 : HcclResult HostCpuUrmaChannel::BuildConnection()
     124              : {
     125            8 :     UbConnBuildContext ctx;
     126            8 :     CHK_RET(PrepareUbConnBuildContext(localEp_, remoteEp_, channelDesc_, ctx));
     127              : 
     128            8 :     Hccl::OpMode opMode = Hccl::OpMode::OPBASE;
     129            8 :     std::unique_ptr<Hccl::HostUbConnection> ubConn = nullptr;
     130            8 :     switch (ctx.protocol) {
     131            0 :         case Hccl::LinkProtocol::UB_TP:
     132            0 :             EXCEPTION_CATCH(
     133              :                 ubConn = std::make_unique<Hccl::HostUbTpConnection>(
     134              :                     rdmaHandle_, ctx.locAddr, ctx.rmtAddr, opMode, Hccl::HrtUbJfcMode::NORMAL, ctx.qosPre),
     135              :                 return HCCL_E_PTR);
     136            0 :             break;
     137            8 :         case Hccl::LinkProtocol::UB_CTP:
     138            8 :             EXCEPTION_CATCH(
     139              :                 ubConn = std::make_unique<Hccl::HostUbCtpConnection>(
     140              :                     rdmaHandle_, ctx.locAddr, ctx.rmtAddr, opMode, Hccl::HrtUbJfcMode::NORMAL, ctx.qosPre),
     141              :                 return HCCL_E_PTR);
     142            8 :             break;
     143            0 :         default:
     144            0 :             HCCL_ERROR("%s No LinkProtocol protocol[%s] to match", __func__, ctx.protocol.Describe().c_str());
     145            0 :             break;
     146              :     }
     147            8 :     CHK_SMART_PTR_NULL(ubConn);
     148              : 
     149            8 :     commonRes_.connVec.clear();
     150            8 :     commonRes_.connVec.emplace_back(ubConn.get());
     151            8 :     connections_.clear();
     152            8 :     connections_.push_back(std::move(ubConn));
     153              : 
     154            8 :     return HCCL_SUCCESS;
     155            8 : }
     156              : 
     157            8 : HcclResult HostCpuUrmaChannel::BuildUbMemTransport()
     158              : {
     159            8 :     Hccl::BaseMemTransport::LocCntNotifyRes locCntNotifyRes{};
     160            8 :     const Hccl::Socket& socket = *socket_;
     161            8 :     bool isRecvFirst = socket.GetRole() == Hccl::SocketRole::CLIENT ? true : false;
     162              : 
     163            8 :     Hccl::LinkData linkData = BuildDefaultLinkData();
     164            8 :     CHK_RET(EndpointDescPairToLinkData(localEp_, remoteEp_, linkData));
     165              : 
     166              :     // make_unique / make_shared / release 包一层抛异常的宏
     167            8 :     EXCEPTION_CATCH(
     168              :         memTransport_ = std::make_unique<Hccl::UbMemTransport>(
     169              :             commonRes_, attr_, linkData, socket, rdmaHandle_, locCntNotifyRes, isRecvFirst),
     170              :         return HCCL_E_PTR);
     171            8 :     return HCCL_SUCCESS;
     172            8 : }
     173              : 
     174           10 : HcclResult HostCpuUrmaChannel::Init()
     175              : {
     176              :     s32 devLogicId;
     177           10 :     CHK_RET(hrtGetDevice(&devLogicId));
     178           10 :     CHK_RET(hrtGetDevicePhyIdByIndex(static_cast<u32>(devLogicId), devicePhyId_));
     179           10 :     CHK_RET(ParseInputParam());
     180            9 :     if (channelDesc_.role != HCOMM_SOCKET_ROLE_CLIENT) {
     181            1 :         CHK_RET(StartListen());
     182              :     }
     183            8 :     CHK_RET(BuildSocket());
     184            8 :     CHK_RET(BuildConnection());
     185            8 :     CHK_RET(BuildUbMemTransport());
     186              :     // urma函数初始化
     187            8 :     CHK_RET(DlUrmaFunction::GetInstance().DlUrmaFunctionInit());
     188              :     // 获取urma read/write 单个wr的最大传输数据大小
     189            8 :     CHK_RET(HccpRaGetDevBaseAttr(rdmaHandle_, &devBaseAttr_));
     190              : 
     191            8 :     return HCCL_SUCCESS;
     192              : }
     193              : 
     194            0 : HcclResult HostCpuUrmaChannel::GetNotifyNum([[maybe_unused]] uint32_t* notifyNum) const
     195              : {
     196            0 :     HCCL_INFO("[HostCpuUrmaChannel::%s] not supported yet.", __func__);
     197            0 :     return HCCL_SUCCESS;
     198              : }
     199              : 
     200            1 : HcclResult HostCpuUrmaChannel::GetRemoteMems(uint32_t* memNum, CommMem** remoteMem, char*** memInfos)
     201              : {
     202            1 :     return memTransport_->GetRemoteMems(memNum, remoteMem, memInfos);
     203              : }
     204              : 
     205            0 : ChannelStatus HostCpuUrmaChannel::GetStatus()
     206              : {
     207            0 :     memTransport_->SetIsHost();
     208            0 :     ChannelStatus out = Channel::TransportStatusToChannelStatus(memTransport_->GetStatus());
     209            0 :     return out;
     210              : }
     211              : 
     212            0 : HcclResult hcomm::HostCpuUrmaChannel::NotifyRecord([[maybe_unused]] const uint32_t remoteNotifyIdx)
     213              : {
     214            0 :     HCCL_INFO("[HostCpuUrmaChannel::%s] not supported yet.", __func__);
     215            0 :     return HCCL_E_NOT_SUPPORT;
     216              : }
     217              : 
     218            0 : HcclResult hcomm::HostCpuUrmaChannel::NotifyWait(
     219              :     [[maybe_unused]] const uint32_t localNotifyIdx, [[maybe_unused]] const uint32_t timeout)
     220              : {
     221            0 :     HCCL_INFO("[HostCpuUrmaChannel::%s] not supported yet.", __func__);
     222            0 :     return HCCL_E_NOT_SUPPORT;
     223              : }
     224              : 
     225            0 : HcclResult hcomm::HostCpuUrmaChannel::WriteWithNotify(
     226              :     [[maybe_unused]] void* dst, [[maybe_unused]] const void* src, [[maybe_unused]] const uint64_t len,
     227              :     [[maybe_unused]] uint32_t remoteNotifyIdx)
     228              : {
     229            0 :     HCCL_INFO("[HostCpuUrmaChannel::%s] not supported yet.", __func__);
     230            0 :     return HCCL_E_NOT_SUPPORT;
     231              : }
     232              : 
     233            4 : HcclResult HostCpuUrmaChannel::GetLocSeg(const void* addr, const size_t size, u64* seg)
     234              : {
     235            4 :     if (commonRes_.bufferVec.empty()) {
     236            1 :         HCCL_ERROR("[HostCpuUrmaChannel::%s] commonRes_.bufferVec is empty.", __func__);
     237            1 :         return HCCL_E_INTERNAL;
     238              :     }
     239              : 
     240            3 :     bool isAddrInRange = false;
     241            4 :     for (auto& it : commonRes_.bufferVec) {
     242            4 :         CHK_PTR_NULL(it);
     243            3 :         Hccl::Buffer iterBuf(it->GetAddr(), it->GetSize());
     244            3 :         if (iterBuf.Contains(reinterpret_cast<uintptr_t>(addr), size)) {
     245            2 :             auto localUbRmaBuffer = dynamic_cast<Hccl::LocalUbRmaBuffer*>(it);
     246            2 :             CHK_PTR_NULL(localUbRmaBuffer);
     247            1 :             *seg = localUbRmaBuffer->GetTargetSeg();
     248            1 :             isAddrInRange = true;
     249            1 :             break;
     250              :         }
     251            3 :     }
     252              : 
     253            2 :     if (!isAddrInRange) {
     254            1 :         HCCL_ERROR("GetLocSeg addr[%p] size[%llu] is not in commonRes_.bufferVec", addr, size);
     255            1 :         return HCCL_E_INTERNAL;
     256              :     }
     257            1 :     return HCCL_SUCCESS;
     258              : }
     259              : 
     260            4 : HcclResult HostCpuUrmaChannel::GetSplitNum(uint64_t len, uint64_t maxJettyWrDataLen, uint64_t& splitNum)
     261              : {
     262            4 :     if (len == 0 || maxJettyWrDataLen == 0) {
     263            1 :         HCCL_ERROR(
     264              :             "[HostCpuUrmaChannel::%s] invalid len[%llu] or maxJettyWrDataLen[%llu].", __func__, len, maxJettyWrDataLen);
     265            1 :         return HCCL_E_PARA;
     266              :     }
     267            3 :     if ((len % maxJettyWrDataLen) == 0) {
     268            2 :         splitNum = len / maxJettyWrDataLen;
     269              :     } else {
     270            1 :         splitNum = (len / maxJettyWrDataLen) + 1;
     271              :     }
     272            3 :     return HCCL_SUCCESS;
     273              : }
     274              : 
     275            0 : HcclResult HostCpuUrmaChannel::GetLocalAndRemoteSeg(
     276              :     urma_opcode_t opcode, void* dst, const void* src, uint64_t len, u64& localSeg, u64& remoteSeg)
     277              : {
     278            0 :     if (opcode == URMA_OPC_WRITE) {
     279            0 :         CHK_RET(GetLocSeg(src, len, &localSeg));
     280            0 :         CHK_RET(memTransport_->GetRemoteSeg(dst, len, &remoteSeg));
     281            0 :     } else if (opcode == URMA_OPC_READ) {
     282            0 :         CHK_RET(GetLocSeg(dst, len, &localSeg));
     283            0 :         CHK_RET(memTransport_->GetRemoteSeg(src, len, &remoteSeg));
     284              :     }
     285            0 :     return HCCL_SUCCESS;
     286              : }
     287              : 
     288              : constexpr u32 RELAX_ORDER = 1;  // Relax Order
     289              : constexpr u32 STRONG_ORDER = 2; // Strong Order
     290            0 : HcclResult HostCpuUrmaChannel::UrmaPostJettySendWr(urma_opcode_t opcode, void* dst, const void* src, uint64_t len)
     291              : {
     292              :     // 构造urma的wr
     293            0 :     urma_jfs_wr_t urmaWriteWr{};
     294            0 :     urmaWriteWr.opcode = opcode;
     295            0 :     urmaWriteWr.flag.bs.place_order = (fenceFlag_ == true ? STRONG_ORDER : RELAX_ORDER);
     296            0 :     urmaWriteWr.flag.bs.comp_order = 1; // comp_order要一直保持为1,
     297            0 :     urmaWriteWr.flag.bs.fence = (fenceFlag_ == true ? 1 : 0);
     298            0 :     urmaWriteWr.flag.bs.complete_enable = 0;
     299            0 :     urmaWriteWr.flag.bs.inline_flag = 0;
     300            0 :     urmaWriteWr.tjetty = reinterpret_cast<urma_target_jetty_t*>(connections_[0]->GetTJettyVa());
     301            0 :     urmaWriteWr.user_ctx = 0; // 跟ibvs中的wr_id对应
     302            0 :     urmaWriteWr.next = nullptr;
     303              : 
     304              :     //  获取切片数量
     305            0 :     uint64_t splitNum = 0;
     306            0 :     uint64_t maxJettyWrDataLen = (opcode == URMA_OPC_WRITE) ? devBaseAttr_.maxWriteSize : devBaseAttr_.maxReadSize;
     307            0 :     CHK_RET(GetSplitNum(len, maxJettyWrDataLen, splitNum));
     308              : 
     309              :     u64 localSeg;
     310              :     u64 remoteSeg;
     311            0 :     CHK_RET(GetLocalAndRemoteSeg(opcode, dst, src, len, localSeg, remoteSeg));
     312              : 
     313            0 :     uint64_t offset = 0;
     314            0 :     for (uint64_t i = 0; i < splitNum; i++) {
     315            0 :         urma_jfs_wr_t* badWr = nullptr;
     316            0 :         uint64_t chunkLen = std::min(len - offset, maxJettyWrDataLen);
     317              :         // 源地址 数据长度 tseg
     318            0 :         urma_sge_t srclist = {};
     319            0 :         urmaWriteWr.rw.src.sge = &srclist;
     320            0 :         urmaWriteWr.rw.src.sge->addr = reinterpret_cast<uint64_t>(static_cast<char*>(const_cast<void*>(src)) + offset);
     321            0 :         urmaWriteWr.rw.src.sge->len = chunkLen;
     322            0 :         urmaWriteWr.rw.src.sge->tseg = (opcode == URMA_OPC_WRITE) ? reinterpret_cast<urma_target_seg_t*>(localSeg) :
     323              :                                                                     reinterpret_cast<urma_target_seg_t*>(remoteSeg);
     324            0 :         urmaWriteWr.rw.src.num_sge = 1;
     325              : 
     326              :         // 目的地址 数据长度 tseg
     327            0 :         urma_sge_t dstlist = {};
     328            0 :         urmaWriteWr.rw.dst.sge = &dstlist;
     329            0 :         urmaWriteWr.rw.dst.sge->addr = reinterpret_cast<uint64_t>(static_cast<const char*>(dst) + offset); // 远端地址
     330            0 :         urmaWriteWr.rw.dst.sge->len = chunkLen;
     331            0 :         urmaWriteWr.rw.dst.sge->tseg = (opcode == URMA_OPC_WRITE) ? reinterpret_cast<urma_target_seg_t*>(remoteSeg) :
     332              :                                                                     reinterpret_cast<urma_target_seg_t*>(localSeg);
     333            0 :         urmaWriteWr.rw.dst.num_sge = 1;
     334              : 
     335              :         // 只有最后一个wr上报cqe
     336            0 :         if (i == splitNum - 1) {
     337            0 :             urmaWriteWr.flag.bs.complete_enable = 1;
     338            0 :             urmaWriteWr.flag.bs.place_order = STRONG_ORDER; // 最后一个wr设置为strong order
     339              :         }
     340            0 :         CHK_RET(HrtUrmaPostJettySendWr(
     341              :             reinterpret_cast<urma_jetty_t*>(connections_[0]->GetJettyVa()), &urmaWriteWr, &badWr));
     342            0 :         offset += chunkLen;
     343              :     }
     344            0 :     fenceFlag_ = false;
     345            0 :     wqeNum_++;
     346            0 :     HCCL_INFO(
     347              :         "UrmaPostJettySendWr opencode[%u] fenceFlag_[%u] wqeNum_[%u] splitNum[%llu] SUCCESS.", opcode, fenceFlag_,
     348              :         wqeNum_, splitNum);
     349            0 :     return HCCL_SUCCESS;
     350              : }
     351              : 
     352            0 : HcclResult hcomm::HostCpuUrmaChannel::Write(void* dst, const void* src, uint64_t len)
     353              : {
     354            0 :     CHK_RET(UrmaPostJettySendWr(URMA_OPC_WRITE, dst, src, len));
     355            0 :     return HCCL_SUCCESS;
     356              : }
     357              : 
     358            0 : HcclResult hcomm::HostCpuUrmaChannel::Read(void* dst, const void* src, uint64_t len)
     359              : {
     360            0 :     CHK_RET(UrmaPostJettySendWr(URMA_OPC_READ, dst, src, len));
     361            0 :     return HCCL_SUCCESS;
     362              : }
     363              : 
     364            2 : HcclResult hcomm::HostCpuUrmaChannel::ChannelFence()
     365              : {
     366            2 :     std::lock_guard<std::mutex> lock(fenceMutex_);
     367            2 :     HCCL_INFO("[HostCpuUrmaChannel::%s] start, wqeNum_ = %u va[%llu]", __func__, wqeNum_, connections_[0]->GetCqVa());
     368            2 :     CHK_PRT_RET(
     369              :         wqeNum_ == 0, HCCL_INFO("[HostCpuUrmaChannel::%s] no need to fence since no wqeNum[%u].", __func__),
     370              :         HCCL_SUCCESS);
     371            1 :     std::vector<urma_cr_t> wc(wqeNum_);
     372              : 
     373              :     auto timeout = std::chrono::milliseconds(
     374            1 :         static_cast<uint64_t>(Hccl::EnvConfig::GetInstance().GetRtsConfig().GetExecTimeOut())
     375            1 :         * 1000ULL); // 乘1000转为毫秒
     376            1 :     auto startTime = std::chrono::steady_clock::now();
     377              :     while (true) {
     378            1 :         auto actualNum = HrtUrmaPollJfc(reinterpret_cast<urma_jfc_t*>(connections_[0]->GetCqVa()), wqeNum_, wc.data());
     379            1 :         if (actualNum < 0) {
     380            1 :             HCCL_ERROR("[HostCpuUrmaChannel::%s] urma_poll_jfc failed. actualNum=%d", __func__, actualNum);
     381            1 :             return HCCL_E_NETWORK;
     382              :         }
     383              : 
     384            0 :         uint32_t actualNum32 = static_cast<uint32_t>(actualNum);
     385            0 :         if (actualNum32 > wqeNum_) {
     386            0 :             HCCL_ERROR(
     387              :                 "[HostCpuUrmaChannel::%s] urma_poll_jfc polled more completions (%u) than expected (%u).", __func__,
     388              :                 actualNum32, wqeNum_);
     389            0 :             return HCCL_E_INTERNAL;
     390            0 :         } else if (actualNum32 > 0) {
     391            0 :             for (uint32_t i = 0; i < actualNum32; i++) {
     392            0 :                 if (wc[i].status != URMA_CR_SUCCESS) {
     393            0 :                     HCCL_ERROR(
     394              :                         "[HostCpuUrmaChannel::%s] urma_poll_jfc error. wc[%u] status:%d", __func__, i, wc[i].status);
     395            0 :                     return HCCL_E_NETWORK;
     396              :                 }
     397              :             }
     398            0 :             wqeNum_ -= actualNum32; // 减去已完成的数量,继续等待剩余的完成
     399            0 :             if (wqeNum_ == 0) {
     400            0 :                 break; // 所有的wqe都已完成,退出循环
     401              :             }
     402              :         }
     403              : 
     404            0 :         if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
     405            0 :             HCCL_ERROR("[HostCpuUrmaChannel::%s] call urma_poll_jfc timeout.", __func__);
     406            0 :             return HCCL_E_TIMEOUT;
     407              :         }
     408            0 :     }
     409              : 
     410            0 :     wqeNum_ = 0; // 所有wqe都已完成,重置计算器
     411            0 :     fenceFlag_ = true;
     412            0 :     return HCCL_SUCCESS;
     413            2 : }
     414              : 
     415            1 : HcclResult hcomm::HostCpuUrmaChannel::Clean()
     416              : {
     417            1 :     HCCL_INFO("[HostCpuUrmaChannel::%s] not supported yet.", __func__);
     418            1 :     return HCCL_E_NOT_SUPPORT;
     419              : }
     420              : 
     421            1 : HcclResult hcomm::HostCpuUrmaChannel::Resume()
     422              : {
     423            1 :     HCCL_INFO("[HostCpuUrmaChannel::%s] not supported yet.", __func__);
     424            1 :     return HCCL_E_NOT_SUPPORT;
     425              : }
     426              : 
     427              : } // namespace hcomm
        

Generated by: LCOV version 2.0-1