LCOV - code coverage report
Current view: top level - coll_communicator_mgr/resource_mgr/remote/rank_pairs - channel_manager.cc (source / functions) Coverage Total Hit
Test: coverage.info Lines: 31.4 % 452 142
Test Date: 2026-08-18 17:47:01 Functions: 51.4 % 35 18

            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 "channel_manager.h"
      12              : #include "adapter_rts_common.h"
      13              : #include "log.h"
      14              : #include "comm_configer.h"
      15              : #include "launch_aicpu.h"
      16              : #include "comm_engine_utils.h"
      17              : #include <unordered_set>
      18              : #include <string>
      19              : #include "adapter_prof.h"
      20              : #include "hcom_host_profiling.h"
      21              : 
      22              : namespace hccl {
      23              : 
      24              : constexpr u32 RDMA_NOTIFY_MIN_NUM = 3;
      25              : constexpr u32 NOTIFY_NUM_MAX = 64; // HcclChannelDesc 中 notifynum 的默认限制最大为64
      26              : 
      27          261 : HcclResult ChannelManager::Init(aclrtBinHandle binHandle, u32 userRank, const ManagerCallbacks& callbacks)
      28              : {
      29          261 :     binHandle_ = binHandle;
      30          261 :     userRank_ = userRank;
      31          261 :     callbacks_ = callbacks;
      32          261 :     return HCCL_SUCCESS;
      33              : }
      34              : 
      35          404 : HcclResult ChannelManager::SetChannelCallbacks(const ChannelManagerCallbacks& channelCallbacks)
      36              : {
      37          404 :     channelCallbacks_ = channelCallbacks;
      38          404 :     rankInfoList_ = channelCallbacks_.getRankLists();
      39          404 :     return HCCL_SUCCESS;
      40              : }
      41              : 
      42            6 : std::vector<HcclMemHandle> ChannelManager::CollectMemHandles(const std::vector<HcclChannelDesc>& descs) const
      43              : {
      44            6 :     std::vector<HcclMemHandle> handles;
      45            6 :     std::unordered_set<HcclMemHandle> seen;
      46           15 :     for (const auto& desc : descs) {
      47           18 :         for (uint32_t i = 0; i < desc.memHandleNum; ++i) {
      48            9 :             if (desc.memHandles != nullptr && seen.insert(desc.memHandles[i]).second) {
      49            7 :                 handles.push_back(desc.memHandles[i]);
      50              :             }
      51              :         }
      52              :     }
      53            6 :     return handles;
      54            6 : }
      55              : 
      56            7 : HcclResult ChannelManager::CheckChannelParam(CommEngine engine, const HcclChannelDesc* channelDesc, uint32_t descNum)
      57              : {
      58            7 :     std::unordered_set<HcclChannelDesc, std::hash<HcclChannelDesc>, HcclChannelDescEqual> descSet;
      59              : 
      60           10 :     for (uint32_t descIdx = 0; descIdx < descNum; ++descIdx) {
      61              :         // 检查notifyNum
      62            9 :         CHK_PRT_RET(
      63              :             channelDesc[descIdx].notifyNum > NOTIFY_NUM_MAX,
      64              :             HCCL_ERROR(
      65              :                 "[%s]Channeldesc[%u] invalid notifyNum, notifyNum[%u], max notify num[%u]", __func__, descIdx,
      66              :                 channelDesc[descIdx].notifyNum, NOTIFY_NUM_MAX),
      67              :             HCCL_E_PARA);
      68              :         // 检查HcclChannelDesc是否有重复元素
      69            7 :         CHK_PRT_RET(
      70              :             descSet.find(channelDesc[descIdx]) != descSet.end(),
      71              :             HCCL_ERROR("[%s]Duplicate item found in hcclchanneldesc.", __func__), HCCL_E_PARA);
      72            6 :         descSet.insert(channelDesc[descIdx]);
      73              :         // 检查RemoteRank有效性
      74            6 :         CHK_PRT_RET(
      75              :             channelDesc[descIdx].remoteRank == userRank_,
      76              :             HCCL_ERROR("[%s]Local rank found in channeldesc, userRank_ = %u.", __func__, userRank_), HCCL_E_PARA);
      77              :         // 检查是否有不支持协议
      78            5 :         if (engine == COMM_ENGINE_AIV) {
      79            0 :             CHK_PRT_RET(
      80              :                 channelDesc[descIdx].channelProtocol != COMM_PROTOCOL_HCCS
      81              :                     && channelDesc[descIdx].channelProtocol != COMM_PROTOCOL_HCCS_ONLY
      82              :                     && channelDesc[descIdx].channelProtocol != COMM_PROTOCOL_SIO,
      83              :                 HCCL_ERROR(
      84              :                     "[%s]AIV engine Unsupported protocol[%d] found in channeldesc, protocol: %d.", __func__, descIdx,
      85              :                     channelDesc[descIdx].channelProtocol),
      86              :                 HCCL_E_PARA);
      87              :         } else {
      88            5 :             CHK_PRT_RET(
      89              :                 channelDesc[descIdx].channelProtocol != COMM_PROTOCOL_HCCS
      90              :                     && channelDesc[descIdx].channelProtocol != COMM_PROTOCOL_ROCE
      91              :                     && channelDesc[descIdx].channelProtocol != COMM_PROTOCOL_HCCS_ONLY
      92              :                     && channelDesc[descIdx].channelProtocol != COMM_PROTOCOL_SIO,
      93              :                 HCCL_ERROR(
      94              :                     "[%s]Unsupported protocol[%d] found in channeldesc, protocol: %d.", __func__, descIdx,
      95              :                     channelDesc[descIdx].channelProtocol),
      96              :                 HCCL_E_PARA);
      97              :         }
      98              : 
      99              :         // 检查engine支持情况
     100            4 :         if (engine != COMM_ENGINE_CPU && engine != COMM_ENGINE_CPU_TS && engine != COMM_ENGINE_AICPU
     101            1 :             && engine != COMM_ENGINE_AICPU_TS && engine != COMM_ENGINE_AIV) {
     102            1 :             HCCL_ERROR(
     103              :                 "[%s]Unsupported engine for channel, engine: %s.", __func__,
     104              :                 GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str());
     105            1 :             return HCCL_E_PARA;
     106              :         }
     107              :     }
     108            1 :     return HCCL_SUCCESS;
     109            7 : }
     110              : 
     111           28 : static std::string BuildChannelKey(const std::string& tag, CommEngine engine, const HcclChannelDesc& channelDesc)
     112              : {
     113           56 :     std::string key = tag + ":" + std::to_string(engine) + ":" + std::to_string(channelDesc.remoteRank) + ":"
     114           84 :                       + std::to_string(channelDesc.channelProtocol);
     115              :     // 将 memHandles 纳入 key,使得上层更换交换内存时会强制新建 channel,避免复用旧 channel 导致远端地址错配
     116           28 :     if (channelDesc.memHandleNum > 0 && channelDesc.memHandles != nullptr) {
     117           22 :         for (uint32_t i = 0; i < channelDesc.memHandleNum; i++) {
     118           13 :             key += ":" + std::to_string(reinterpret_cast<uintptr_t>(channelDesc.memHandles[i]));
     119              :         }
     120            9 :     } else {
     121           19 :         key += ":0";
     122              :     }
     123           28 :     return key;
     124            0 : }
     125              : 
     126           18 : HcclResult ChannelManager::RegisterHandle(
     127              :     const std::string& tag, CommEngine engine, const HcclChannelDesc& channelDesc, ChannelHandle channelHandle)
     128              : {
     129           18 :     std::string channelKey = BuildChannelKey(tag, engine, channelDesc);
     130              : 
     131           18 :     CHK_PRT_RET(
     132              :         (channelHandleMap_.find(channelKey) != channelHandleMap_.end()),
     133              :         HCCL_ERROR(
     134              :             "[%s]Channel already exists, tag[%s], engine[%s], remoteRank[%d], channelProtocol[%d].", __func__,
     135              :             tag.c_str(), GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), channelDesc.remoteRank,
     136              :             channelDesc.channelProtocol),
     137              :         HCCL_E_PARA);
     138           17 :     channelHandleMap_[channelKey] = channelHandle;
     139           17 :     keyMap_[channelHandle] = channelKey;
     140           17 :     engineMap_[channelHandle] = engine;
     141           17 :     HCCL_INFO("[%s]Register channel handle[%llu], channelKey[%s]", __func__, channelHandle, channelKey.c_str());
     142           17 :     return HCCL_SUCCESS;
     143           18 : }
     144              : 
     145            8 : HcclResult ChannelManager::PrepareHandleArray(
     146              :     const std::string& tag, CommEngine engine, const HcclChannelDesc* channelDesc, uint32_t descNum,
     147              :     ChannelHandle* channelHandleArray, std::vector<HcclChannelDesc>& needCreateDescs,
     148              :     std::vector<uint32_t>& needCreateIndices)
     149              : {
     150            8 :     needCreateDescs.clear();
     151            8 :     needCreateIndices.clear();
     152              : 
     153           18 :     for (uint32_t descIdx = 0; descIdx < descNum; descIdx++) {
     154              :         // 组合channelKey
     155           10 :         std::string channelKey = BuildChannelKey(tag, engine, channelDesc[descIdx]);
     156           10 :         if (channelHandleMap_.find(channelKey) != channelHandleMap_.end()) {
     157            4 :             channelHandleArray[descIdx] = channelHandleMap_[channelKey];
     158            4 :             continue;
     159              :         }
     160            6 :         channelHandleArray[descIdx] = 0;
     161            6 :         needCreateDescs.push_back(channelDesc[descIdx]);
     162            6 :         needCreateIndices.push_back(descIdx);
     163           10 :     }
     164              : 
     165            8 :     return HCCL_SUCCESS;
     166              : }
     167              : 
     168            4 : HcclResult ChannelManager::IsChannelExist(ChannelHandle channel)
     169              : {
     170            4 :     CHK_PRT_RET(
     171              :         (keyMap_.find(channel) == keyMap_.end()), HCCL_ERROR("[%s]ChannelHandle is not exist.", __func__), HCCL_E_PARA);
     172            1 :     HCCL_INFO(
     173              :         "[%s]ChannelHandle exist, ChannelHandle[%llu], channelKey[%s]", __func__, channel, keyMap_[channel].c_str());
     174            1 :     return HCCL_SUCCESS;
     175              : }
     176              : 
     177            6 : HcclResult ChannelManager::UnregisterHandle(ChannelHandle channel)
     178              : {
     179            6 :     CHK_PRT_RET(
     180              :         (keyMap_.find(channel) == keyMap_.end()), HCCL_ERROR("[%s]ChannelHandle is not exist.", __func__), HCCL_E_PARA);
     181              : 
     182            4 :     channelHandleMap_.erase(keyMap_[channel]);
     183            4 :     keyMap_.erase(channel);
     184            4 :     if (engineMap_[channel] == COMM_ENGINE_AICPU || engineMap_[channel] == COMM_ENGINE_AICPU_TS) {
     185            0 :         channelD2HMap_.erase(channel);
     186              :     }
     187            4 :     engineMap_.erase(channel);
     188              : 
     189            4 :     HCCL_INFO("[%s]Unregister channel handle success.", __func__);
     190            4 :     return HCCL_SUCCESS;
     191              : }
     192              : 
     193            6 : HcclResult ChannelManager::RegisterHandleHDPair(ChannelHandle deviceChannelHandle, ChannelHandle hostChannelHandle)
     194              : {
     195            6 :     CHK_PRT_RET(
     196              :         (deviceChannelHandle == 0 || hostChannelHandle == 0), HCCL_ERROR("[%s]ChannelHandle is 0.", __func__),
     197              :         HCCL_E_PARA);
     198            4 :     CHK_PRT_RET(
     199              :         (channelD2HMap_.find(deviceChannelHandle) != channelD2HMap_.end()),
     200              :         HCCL_ERROR("[%s]deviceChannelHandle has existed in channelD2HMap_.", __func__), HCCL_E_PARA);
     201              : 
     202            3 :     channelD2HMap_[deviceChannelHandle] = hostChannelHandle;
     203            3 :     return HCCL_SUCCESS;
     204              : }
     205              : 
     206            3 : HcclResult ChannelManager::GetHostChannel(ChannelHandle channel, ChannelHandle& hostChannel)
     207              : {
     208            3 :     if (engineMap_[channel] == COMM_ENGINE_AICPU || engineMap_[channel] == COMM_ENGINE_AICPU_TS) {
     209            2 :         CHK_PRT_RET(
     210              :             (channelD2HMap_.find(channel) == channelD2HMap_.end()),
     211              :             HCCL_ERROR("[%s]device channel handle has not existed in channelD2HMap_.", __func__), HCCL_E_PARA);
     212            1 :         hostChannel = channelD2HMap_[channel];
     213              :     } else {
     214            1 :         hostChannel = channel;
     215              :     }
     216            2 :     return HCCL_SUCCESS;
     217              : }
     218              : 
     219            4 : void ChannelManager::ClearOpTransportResponseLinks(OpCommTransport& opTransportResponse)
     220              : {
     221            8 :     for (auto& levelNSubCommTransport : opTransportResponse) {
     222            8 :         for (auto& singleSubCommTransport : levelNSubCommTransport) {
     223            4 :             u32 size = singleSubCommTransport.transportRequests.size();
     224            4 :             singleSubCommTransport.links.resize(size, nullptr);
     225            4 :             singleSubCommTransport.status.resize(size, TransportStatus::INIT);
     226            4 :             HCCL_INFO("[%s] size[%u], linksSize[%zu]", __func__, size, singleSubCommTransport.links.size());
     227              :         }
     228              :     }
     229            4 : }
     230              : 
     231            3 : HcclResult ChannelManager::CheckNotifyOrQPMaxNum(u64& existNum, const u64& MaxNum, const bool& isNotifyRes)
     232              : {
     233            3 :     std::string resType = isNotifyRes ? "Notify" : "QP";
     234            3 :     if (existNum + 1 > MaxNum) {
     235            1 :         HCCL_ERROR(
     236              :             "[%s]%s resources are insufficient, existNum[%llu], MaxNum is [%llu]", __func__, resType.c_str(), existNum,
     237              :             MaxNum);
     238            1 :         return HCCL_E_INTERNAL;
     239              :     }
     240            2 :     HCCL_DEBUG(
     241              :         "[%s]%s resources are sufficient, existNum[%llu], MaxNum is [%llu]", __func__, resType.c_str(), existNum,
     242              :         MaxNum);
     243            2 :     return HCCL_SUCCESS;
     244            3 : }
     245              : 
     246            0 : HcclResult ChannelManager::CreateWorkSpace(u64 size, DeviceMem& buffer) const
     247              : {
     248            0 :     CHK_PRT_RET(
     249              :         size == 0, HCCL_INFO("[Create][WorkSpace]work space size is zero. not need to malloc memory"), HCCL_SUCCESS);
     250              : 
     251              :     CHK_PRT_RET(
     252              :         (size > ULONG_MAX), HCCL_ERROR("[Create][WorkSpace]work space size is greater than %llu", ULONG_MAX),
     253              :         HCCL_E_PARA);
     254              : 
     255            0 :     u64 memSize = size;
     256            0 :     buffer = DeviceMem::alloc(memSize);
     257            0 :     CHK_PRT_RET(
     258              :         size > 0 && !buffer,
     259              :         HCCL_ERROR(
     260              :             "[Create][WorkSpace]Create work space size[%llu] fail,"
     261              :             "please check workspace size.",
     262              :             size),
     263              :         HCCL_E_PTR);
     264            0 :     CHK_RET(hrtMemSet(buffer.ptr(), size, size));
     265            0 :     return HCCL_SUCCESS;
     266              : }
     267              : 
     268            0 : HcclResult ChannelManager::AllocAndClearHostMem(u64 size, std::shared_ptr<HostMem>& bufferPtr) const
     269              : {
     270            0 :     CHK_PRT_RET(
     271              :         size == 0,
     272              :         HCCL_INFO("[ChannelManager][AllocAndClearHostMem] host memory size is zero. not need to malloc memory"),
     273              :         HCCL_SUCCESS);
     274              : 
     275              :     CHK_PRT_RET(
     276              :         (size > ULONG_MAX),
     277              :         HCCL_ERROR("[ChannelManager][AllocAndClearHostMem] host memory size is greater than %llu", ULONG_MAX),
     278              :         HCCL_E_PARA);
     279              : 
     280            0 :     HostMem tmpBuffer = HostMem::alloc(size);
     281            0 :     EXCEPTION_CATCH((bufferPtr = std::make_shared<HostMem>(std::move(tmpBuffer))), return HCCL_E_PTR);
     282              : 
     283            0 :     CHK_PRT_RET(
     284              :         size > 0 && !bufferPtr.get()->ptr(),
     285              :         HCCL_ERROR(
     286              :             "[ChannelManager][AllocAndClearHostMem]host memory space size[%llu] fail,"
     287              :             "please check workspace size.",
     288              :             size),
     289              :         HCCL_E_PTR);
     290            0 :     CHK_SAFETY_FUNC_RET(memset_s(bufferPtr.get()->ptr(), size, 0, size));
     291            0 :     return HCCL_SUCCESS;
     292            0 : }
     293              : 
     294              : template <typename T>
     295            0 : HcclResult ChannelManager::CopyVectorToDeviceMem(const u64 len, DeviceMem& dstDeviceMem, const std::vector<T>& srcVec)
     296              : {
     297            0 :     CHK_PRT_RET(
     298              :         len == 0, HCCL_INFO("[ChannelManager][CopyVectorToDeviceMem] space size is zero. not need to malloc memory"),
     299              :         HCCL_SUCCESS);
     300              : 
     301              :     CHK_PRT_RET(
     302              :         (len > ULONG_MAX),
     303              :         HCCL_ERROR("[ChannelManager][CopyVectorToDeviceMem] space size is greater than %llu", ULONG_MAX), HCCL_E_PARA);
     304              : 
     305            0 :     CHK_RET(CreateWorkSpace(len, dstDeviceMem));
     306            0 :     std::shared_ptr<HostMem> srcHostMem;
     307            0 :     CHK_RET(AllocAndClearHostMem(len, srcHostMem));
     308            0 :     std::copy(srcVec.begin(), srcVec.end(), static_cast<T*>(srcHostMem.get()->ptr()));
     309            0 :     CHK_RET(hrtMemSyncCopy(
     310              :         dstDeviceMem.ptr(), len, srcHostMem.get()->ptr(), len, HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE));
     311            0 :     return HCCL_SUCCESS;
     312            0 : }
     313              : 
     314            3 : OpCommTransport ChannelManager::BuildChannelRequests(const std::vector<HcclChannelDesc>& descs)
     315              : {
     316            3 :     OpCommTransport opCommTransport;
     317            3 :     LevelNSubCommTransport level0Transport;
     318            3 :     SingleSubCommTransport commTransport;
     319              : 
     320            7 :     for (auto desc : descs) {
     321            4 :         TransportRequest tmpTransport;
     322            4 :         tmpTransport.isValid = true;
     323            4 :         tmpTransport.localUserRank = userRank_;
     324            4 :         tmpTransport.remoteUserRank = desc.remoteRank;
     325            4 :         tmpTransport.notifyNum = desc.notifyNum;
     326            4 :         tmpTransport.inputMemType = TransportMemType::CCL_INPUT;
     327            4 :         tmpTransport.outputMemType = TransportMemType::CCL_OUTPUT;
     328            4 :         tmpTransport.isUsedRdma = (desc.channelProtocol == CommProtocol::COMM_PROTOCOL_ROCE);
     329            4 :         TransportLinkType linkType = TransportLinkType::RESERVED;
     330            4 :         if (desc.channelProtocol == CommProtocol::COMM_PROTOCOL_HCCS_ONLY) {
     331            0 :             linkType = TransportLinkType::HCCS;
     332            4 :         } else if (desc.channelProtocol == CommProtocol::COMM_PROTOCOL_SIO) {
     333            1 :             linkType = TransportLinkType::SIO;
     334              :         }
     335            4 :         tmpTransport.linkType = linkType;
     336            4 :         commTransport.transportRequests.push_back(tmpTransport);
     337              :     }
     338              : 
     339            3 :     level0Transport.push_back(commTransport);
     340            3 :     opCommTransport.push_back(level0Transport);
     341            3 :     ClearOpTransportResponseLinks(opCommTransport);
     342              : 
     343            3 :     return opCommTransport;
     344            3 : }
     345              : 
     346            0 : HcclResult ChannelManager::ParseChannelRemoteDataToMem(
     347              :     const OpCommTransport& opTransportResponse, HcclIndOpChannelRemoteResV3& channelParam)
     348              : {
     349            0 :     uint32_t level0 = 0;
     350            0 :     auto& singleSubCommTransport = opTransportResponse[level0][level0];
     351            0 :     CHK_PRT_RET(
     352              :         channelParam.listNum == 0, HCCL_ERROR("[%s]invalid listNum, listNum[%u]", __func__, channelParam.listNum),
     353              :         HCCL_E_PARA);
     354            0 :     CHK_PRT_RET(
     355              :         (channelParam.listNum != singleSubCommTransport.links.size()),
     356              :         HCCL_ERROR(
     357              :             "[%s]invalid listNum, listNum[%u] but links size is [%zu]", __func__, channelParam.listNum,
     358              :             singleSubCommTransport.links.size()),
     359              :         HCCL_E_PARA);
     360              :     // 分配 HcclIndOpChannelRemoteResV2 内存,需要手动释放
     361              :     channelParam.remoteResV2
     362            0 :         = static_cast<HcclIndOpChannelRemoteResV2*>(malloc(channelParam.listNum * sizeof(HcclIndOpChannelRemoteResV2)));
     363            0 :     CHK_PRT_RET(
     364              :         channelParam.remoteResV2 == nullptr, HCCL_ERROR("[%s]channelParam.remoteResV2 is null.", __func__),
     365              :         HCCL_E_MEMORY);
     366            0 :     u32 linkIdx = 0;
     367            0 :     for (auto& transportRequest : singleSubCommTransport.transportRequests) {
     368            0 :         auto& tempLink = singleSubCommTransport.links[linkIdx];
     369            0 :         channelParam.remoteResV2[linkIdx].remoteWorldRank = rankInfoList_[transportRequest.remoteUserRank].worldRank;
     370            0 :         channelParam.remoteResV2[linkIdx].remoteRank = transportRequest.remoteUserRank;
     371              :         // transport信息保存(notify、qp)
     372            0 :         if (!transportRequest.isUsedRdma) {
     373              :             // sdma -> P2P
     374            0 :             CHK_RET(BuildOpRemoteChannelP2pResParam(tempLink, channelParam.remoteResV2[linkIdx]));
     375            0 :             channelParam.remoteResV2[linkIdx].channelP2p.qos = hcclQos_;
     376            0 :             HCCL_INFO(
     377              :                 "[ChannelManager] [ParseChannelRemoteDataToMem] hcclQos[%u]",
     378              :                 channelParam.remoteResV2[linkIdx].channelP2p.qos);
     379              :         } else {
     380              :             // rdma -> roce
     381            0 :             CHK_RET(BuildOpRemoteChannelRoceResParam(tempLink, channelParam.remoteResV2[linkIdx]));
     382              :         }
     383            0 :         linkIdx++;
     384              :     }
     385            0 :     return HCCL_SUCCESS;
     386              : }
     387              : 
     388            0 : HcclResult ChannelManager::BuildOpRemoteChannelP2pResParam(const LINK& link, HcclIndOpChannelRemoteResV2& remoteRes)
     389              : {
     390            0 :     remoteRes.isUsedRdma = false;
     391            0 :     HcclChannelP2p& linkp2p = remoteRes.channelP2p;
     392              :     // remoteMem, 独立算子localmem是否需要传待确认
     393            0 :     void* bufferPtr = nullptr;
     394            0 :     CHK_RET(link->GetRemoteMem(UserMemType::INPUT_MEM, &bufferPtr));
     395            0 :     linkp2p.remoteHcclbuffer.addr = reinterpret_cast<void*>(bufferPtr);
     396              :     u64 remotebufferSize;
     397            0 :     CHK_RET(link->GetRemoteMemSize(UserMemType::INPUT_MEM, remotebufferSize));
     398            0 :     linkp2p.remoteHcclbuffer.size = remotebufferSize;
     399              :     // 独立算子远端用户内存,linkp2p.remoteUserMem需要手动释放内存
     400            0 :     CHK_RET(link->GetIndOpRemoteMem(&linkp2p.remoteUserMem, &linkp2p.remoteUserMemCount));
     401            0 :     HCCL_DEBUG("[%s] finish set remoteMem info", __func__);
     402              : 
     403              :     // localnotify & remotenotify
     404            0 :     u64 notifyNum = 0;
     405            0 :     std::vector<HcclSignalInfo> locIpcSignals;
     406            0 :     std::vector<HcclSignalInfo> rmtIpcSignals;
     407            0 :     CHK_RET(link->GetLocalNotify(locIpcSignals));
     408            0 :     CHK_RET(link->GetRemoteNotify(rmtIpcSignals));
     409              : 
     410            0 :     for (size_t i = 0; i < locIpcSignals.size(); i++) {
     411            0 :         linkp2p.localIpcSignal[notifyNum] = locIpcSignals[i];
     412            0 :         linkp2p.remoteIpcSignal[notifyNum] = rmtIpcSignals[i];
     413            0 :         notifyNum++;
     414              :     }
     415            0 :     remoteRes.p2pNotifyNum = link->GetNotifyNum();
     416            0 :     HCCL_DEBUG(
     417              :         "[%s] finish set localnotify & remotenotify info, notifyNum[%llu], p2pNotifyNum[%llu]", __func__, notifyNum,
     418              :         remoteRes.p2pNotifyNum);
     419              :     // transportAttr
     420            0 :     CHK_RET(link->GetTransportAttr(linkp2p.transportAttr));
     421            0 :     HCCL_DEBUG("[%s] finish set RemoteChannelP2pResParam info", __func__);
     422            0 :     return HCCL_SUCCESS;
     423            0 : }
     424              : 
     425            0 : HcclResult ChannelManager::BuildOpRemoteChannelRoceResParam(const LINK& link, HcclIndOpChannelRemoteResV2& remoteRes)
     426              : {
     427            0 :     remoteRes.isUsedRdma = true;
     428            0 :     HcclChannelRoce& linkRoce = remoteRes.channelRoce;
     429              :     // 填充localMem信息到linkRoce中
     430            0 :     CHK_RET(link->GetLocalMemDetails(UserMemType::INPUT_MEM, linkRoce.localHcclbuffer));
     431              :     // 填充remoteMem信息到linkRoce中
     432            0 :     void* bufferPtr = nullptr;
     433            0 :     CHK_RET(link->GetRemoteMem(UserMemType::INPUT_MEM, &bufferPtr));
     434            0 :     linkRoce.remoteHcclbuffer.addr = reinterpret_cast<u64>(bufferPtr);
     435            0 :     CHK_RET(link->GetRemoteMemKey(UserMemType::INPUT_MEM, &(linkRoce.remoteHcclbuffer.key)));
     436            0 :     CHK_RET(link->GetRemoteMemSize(UserMemType::INPUT_MEM, linkRoce.remoteHcclbuffer.size));
     437              :     // 独立算子远端用户内存,linkRoce.remoteUserHostMem和remoteUserDeviceMem需要手动释放内存
     438            0 :     CHK_RET(link->GetIndOpRemoteMemDetails(
     439              :         &linkRoce.remoteUserHostMem, &linkRoce.remoteUserHostMemCount, HcclMemType::HCCL_MEM_TYPE_HOST));
     440            0 :     CHK_RET(link->GetIndOpRemoteMemDetails(
     441              :         &linkRoce.remoteUserDeviceMem, &linkRoce.remoteUserHostMemCount, HcclMemType::HCCL_MEM_TYPE_DEVICE));
     442            0 :     HCCL_DEBUG("[%s] finish set remoteMem info", __func__);
     443              : 
     444              :     // 填充notifyValue和notifyValueKey信息到linkRoce中
     445            0 :     std::vector<AddrKey> notifyValueAddrKey;
     446            0 :     CHK_RET(link->GetLocalNotifyValueAddrKey(notifyValueAddrKey));
     447            0 :     linkRoce.notifyValue = notifyValueAddrKey[0].addr;
     448            0 :     linkRoce.notifyValueKey = notifyValueAddrKey[0].key;
     449              : 
     450              :     // 填充QP信息到linkRoce中
     451            0 :     std::vector<HcclQpInfoV2> aiQpInfos;
     452            0 :     CHK_RET(link->GetAiQpInfo(aiQpInfos));
     453            0 :     u32 qpNum = aiQpInfos.size();
     454            0 :     if (qpNum > RDMA_QP_MAX_NUM || qpNum < 1) {
     455            0 :         return HCCL_E_INTERNAL;
     456              :     }
     457            0 :     std::copy_n(aiQpInfos.begin(), qpNum, linkRoce.QpInfo);
     458            0 :     linkRoce.qpsPerConnection = qpNum - static_cast<u32>(qpNum > 1); // 多QP数量或单QP模式
     459              : 
     460              :     // 填充localNotify和remoteNotify信息到linkRoce中
     461            0 :     std::vector<AddrKey> notifyAddrKey;
     462            0 :     std::vector<HcclSignalInfo> signalInfos;
     463            0 :     CHK_RET(link->GetLocalRdmaNotify(signalInfos));
     464            0 :     CHK_RET(link->GetRemoteRdmaNotifyAddrKey(notifyAddrKey));
     465            0 :     if ((signalInfos.size() != notifyAddrKey.size()) || (signalInfos.size() < RDMA_NOTIFY_MIN_NUM)
     466            0 :         || (signalInfos.size() > RDMA_NOTIFY_MAX_NUM) || (notifyAddrKey.size() < RDMA_NOTIFY_MIN_NUM)
     467            0 :         || (notifyAddrKey.size() > RDMA_NOTIFY_MAX_NUM)
     468            0 :         || ((signalInfos.size() - RDMA_NOTIFY_MIN_NUM) % linkRoce.qpsPerConnection) != 0
     469            0 :         || ((notifyAddrKey.size() - RDMA_NOTIFY_MIN_NUM) % linkRoce.qpsPerConnection) != 0) {
     470            0 :         return HCCL_E_INTERNAL;
     471              :     }
     472            0 :     u64 notifyNum = (notifyAddrKey.size() - RDMA_NOTIFY_MIN_NUM) / linkRoce.qpsPerConnection
     473            0 :                     - static_cast<u32>(linkRoce.qpsPerConnection > 1);
     474            0 :     linkRoce.singleQPNotifyNum = notifyNum;
     475              : 
     476            0 :     u64 len = signalInfos.size() * sizeof(HcclSignalInfo);
     477            0 :     DeviceMem localNotifyListMem;
     478            0 :     CHK_RET(CopyVectorToDeviceMem(len, localNotifyListMem, signalInfos));
     479            0 :     linkRoce.localNotifyList = reinterpret_cast<u64>(localNotifyListMem.ptr());
     480            0 :     channelParamMemList_.emplace_back(std::move(localNotifyListMem));
     481              : 
     482            0 :     len = notifyAddrKey.size() * sizeof(AddrKey);
     483            0 :     DeviceMem remoteNotifyListMem;
     484            0 :     CHK_RET(CopyVectorToDeviceMem(len, remoteNotifyListMem, notifyAddrKey));
     485            0 :     linkRoce.remoteNotifyList = reinterpret_cast<u64>(remoteNotifyListMem.ptr());
     486            0 :     channelParamMemList_.emplace_back(std::move(remoteNotifyListMem));
     487              : 
     488            0 :     remoteRes.roceNotifyNum = linkRoce.singleQPNotifyNum;
     489            0 :     remoteRes.qpNum = linkRoce.qpsPerConnection;
     490              : 
     491            0 :     return HCCL_SUCCESS;
     492            0 : }
     493              : 
     494            0 : HcclResult ChannelManager::DeepCopyH2DchannelParam(
     495              :     const HcclIndOpChannelRemoteResV3& hostChannelParam, HcclIndOpChannelRemoteResV3& deviceChannelParam)
     496              : {
     497            0 :     deviceChannelParam = hostChannelParam;
     498              :     // 拷贝remoteResV2
     499              : 
     500            0 :     if (hostChannelParam.remoteResV2 != nullptr && hostChannelParam.listNum > 0) {
     501              :         // 为设备端的remoteResV2数组分配内存(注意:这个数组存放的是HcclIndOpChannelRemoteResV2结构体)
     502            0 :         size_t remoteResV2ArraySize = sizeof(HcclIndOpChannelRemoteResV2) * hostChannelParam.listNum;
     503            0 :         std::shared_ptr<DeviceMem> deviceRemoteResV2Array;
     504            0 :         EXCEPTION_CATCH(
     505              :             (deviceRemoteResV2Array = std::make_shared<DeviceMem>(DeviceMem::alloc(remoteResV2ArraySize))),
     506              :             return HCCL_E_PTR);
     507              : 
     508              :         // 为每个数组元素进行深度拷贝,并保存设备内存和主机结构体(指针已调整)
     509            0 :         std::vector<DeviceMem> elementMemories; // 保存每个元素分配的设备内存(包括内部指针数据)
     510            0 :         std::vector<HcclIndOpChannelRemoteResV2> hostRemoteResV2Array(hostChannelParam.listNum);
     511              : 
     512            0 :         for (uint32_t i = 0; i < hostChannelParam.listNum; ++i) {
     513            0 :             HcclIndOpChannelRemoteResV2 hostElement = hostChannelParam.remoteResV2[i];
     514            0 :             HcclIndOpChannelRemoteResV2 deviceElement;
     515              :             // 深度拷贝一个元素到设备内存,并返回设备内存中的结构体布局(host端)
     516            0 :             CHK_RET(DeepCopyH2DChannelRemoteResV2(hostElement, deviceElement));
     517              :             // 保存调整后的主机端结构体(其指针指向设备内存)
     518            0 :             hostRemoteResV2Array[i] = deviceElement;
     519              :         }
     520              : 
     521              :         // 将主机端的结构体数组(指针已调整)拷贝到设备内存数组
     522            0 :         CHK_RET(hrtMemSyncCopy(
     523              :             deviceRemoteResV2Array.get()->ptr(), remoteResV2ArraySize, hostRemoteResV2Array.data(),
     524              :             remoteResV2ArraySize, HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE));
     525              : 
     526              :         // 更新设备端参数中的remoteResV2指针
     527              :         deviceChannelParam.remoteResV2
     528            0 :             = reinterpret_cast<HcclIndOpChannelRemoteResV2*>(deviceRemoteResV2Array.get()->ptr());
     529            0 :         channelParamMemVector_.push_back(std::move(deviceRemoteResV2Array));
     530            0 :     } else {
     531            0 :         HCCL_ERROR("[%s]invalid hostChannelParam", __func__);
     532            0 :         return HCCL_E_INTERNAL;
     533              :     }
     534            0 :     return HCCL_SUCCESS;
     535              : }
     536              : 
     537            0 : HcclResult ChannelManager::DeepCopyH2DChannelRemoteResV2(
     538              :     const HcclIndOpChannelRemoteResV2& hostRemoteResV2, HcclIndOpChannelRemoteResV2& deviceRemoteResV2)
     539              : {
     540              :     // 复制基本成员
     541            0 :     deviceRemoteResV2 = hostRemoteResV2;
     542              :     // 根据通信类型处理不同的通道
     543            0 :     if (hostRemoteResV2.isUsedRdma) {
     544              :         // 处理RoCE通道
     545            0 :         CHK_RET(DeepCopyH2DChannelRoce(hostRemoteResV2.channelRoce, deviceRemoteResV2.channelRoce));
     546              :     } else {
     547              :         // 处理P2P通道
     548            0 :         CHK_RET(DeepCopyH2DChannelP2p(hostRemoteResV2.channelP2p, deviceRemoteResV2.channelP2p));
     549              :     }
     550            0 :     return HCCL_SUCCESS;
     551              : }
     552              : 
     553              : HcclResult
     554            0 : ChannelManager::DeepCopyH2DChannelRoce(const HcclChannelRoce& hostChannelRoce, HcclChannelRoce& deviceChannelRoce)
     555              : {
     556              :     // 复制基本成员
     557            0 :     deviceChannelRoce = hostChannelRoce;
     558              :     // 处理remoteUserHostMem
     559            0 :     if (hostChannelRoce.remoteUserHostMem != nullptr && hostChannelRoce.remoteUserHostMemCount > 0) {
     560            0 :         size_t remoteUserHostMemSize = hostChannelRoce.remoteUserHostMemCount * sizeof(MemDetails);
     561            0 :         std::shared_ptr<DeviceMem> deviceMem;
     562            0 :         EXCEPTION_CATCH(
     563              :             (deviceMem = std::make_shared<DeviceMem>(DeviceMem::alloc(remoteUserHostMemSize))), return HCCL_E_PTR);
     564            0 :         CHK_RET(hrtMemSyncCopy(
     565              :             deviceMem.get()->ptr(), remoteUserHostMemSize, hostChannelRoce.remoteUserHostMem, remoteUserHostMemSize,
     566              :             HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE));
     567            0 :         deviceChannelRoce.remoteUserHostMem = reinterpret_cast<MemDetails*>(deviceMem.get()->ptr());
     568            0 :         channelParamMemVector_.push_back(std::move(deviceMem));
     569            0 :     } else {
     570            0 :         deviceChannelRoce.remoteUserHostMem = nullptr;
     571              :     }
     572              :     // 处理remoteUserDeviceMem
     573            0 :     if (hostChannelRoce.remoteUserDeviceMem != nullptr && hostChannelRoce.remoteUserDeviceMemCount > 0) {
     574            0 :         size_t remoteUserDeviceMemSize = hostChannelRoce.remoteUserDeviceMemCount * sizeof(MemDetails);
     575            0 :         std::shared_ptr<DeviceMem> deviceMem;
     576            0 :         EXCEPTION_CATCH(
     577              :             (deviceMem = std::make_shared<DeviceMem>(DeviceMem::alloc(remoteUserDeviceMemSize))), return HCCL_E_PTR);
     578            0 :         CHK_RET(hrtMemSyncCopy(
     579              :             deviceMem.get()->ptr(), remoteUserDeviceMemSize, hostChannelRoce.remoteUserDeviceMem,
     580              :             remoteUserDeviceMemSize, HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE));
     581            0 :         deviceChannelRoce.remoteUserDeviceMem = reinterpret_cast<MemDetails*>(deviceMem.get()->ptr());
     582            0 :         channelParamMemVector_.push_back(std::move(deviceMem));
     583            0 :     } else {
     584            0 :         deviceChannelRoce.remoteUserDeviceMem = nullptr;
     585              :     }
     586              : 
     587            0 :     return HCCL_SUCCESS;
     588              : }
     589              : 
     590            0 : HcclResult ChannelManager::DeepCopyH2DChannelP2p(const HcclChannelP2p& hostChannelP2p, HcclChannelP2p& deviceChannelP2p)
     591              : {
     592              :     // 复制基本成员
     593            0 :     deviceChannelP2p = hostChannelP2p;
     594              :     // 处理remoteUserMem
     595            0 :     if (hostChannelP2p.remoteUserMem != nullptr && hostChannelP2p.remoteUserMemCount > 0) {
     596            0 :         size_t remoteUserMemSize = hostChannelP2p.remoteUserMemCount * sizeof(HcclMem);
     597            0 :         std::shared_ptr<DeviceMem> deviceMem;
     598            0 :         EXCEPTION_CATCH(
     599              :             (deviceMem = std::make_shared<DeviceMem>(DeviceMem::alloc(remoteUserMemSize))), return HCCL_E_PTR);
     600            0 :         CHK_RET(hrtMemSyncCopy(
     601              :             deviceMem.get()->ptr(), remoteUserMemSize, hostChannelP2p.remoteUserMem, remoteUserMemSize,
     602              :             HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE));
     603            0 :         deviceChannelP2p.remoteUserMem = reinterpret_cast<HcclMem*>(deviceMem.get()->ptr());
     604            0 :         channelParamMemVector_.push_back(std::move(deviceMem));
     605            0 :     } else {
     606            0 :         deviceChannelP2p.remoteUserMem = nullptr;
     607              :     }
     608            0 :     return HCCL_SUCCESS;
     609              : }
     610              : 
     611            0 : HcclResult ChannelManager::ReleaseChannelParam(HcclIndOpChannelRemoteResV3& channelParam)
     612              : {
     613              :     // 释放remoteResV2
     614            0 :     if (channelParam.remoteResV2 != nullptr) {
     615            0 :         for (uint32_t i = 0; i < channelParam.listNum; ++i) {
     616            0 :             HcclIndOpChannelRemoteResV2& remoteRes = channelParam.remoteResV2[i];
     617            0 :             if (remoteRes.isUsedRdma) {
     618            0 :                 if (remoteRes.channelRoce.remoteUserHostMem != nullptr) {
     619            0 :                     free(remoteRes.channelRoce.remoteUserHostMem);
     620              :                 }
     621            0 :                 if (remoteRes.channelRoce.remoteUserDeviceMem != nullptr) {
     622            0 :                     free(remoteRes.channelRoce.remoteUserDeviceMem);
     623              :                 }
     624              :             } else {
     625            0 :                 if (remoteRes.channelP2p.remoteUserMem != nullptr) {
     626            0 :                     free(remoteRes.channelP2p.remoteUserMem);
     627              :                 }
     628              :             }
     629              :         }
     630              :     }
     631            0 :     free(channelParam.remoteResV2);
     632            0 :     channelParam.remoteResV2 = nullptr;
     633              : 
     634              :     // 将kernel下发时临时分配的deviceMem一起销毁
     635            0 :     channelParamMemVector_.clear();
     636            0 :     channelParamMemList_.clear();
     637            0 :     return HCCL_SUCCESS;
     638              : }
     639              : 
     640            0 : HcclResult ChannelManager::AicpuChannelInit(
     641              :     const std::string& commId, const std::string& tag, CommEngine engine, const OpCommTransport& opTransportResponse,
     642              :     ChannelHandle* channelList, uint32_t listNum)
     643              : {
     644            0 :     HcclIndOpChannelRemoteResV3 channelParam{};
     645            0 :     CHK_SAFETY_FUNC_RET(memset_s(&channelParam, sizeof(channelParam), 0, sizeof(channelParam)));
     646            0 :     uint64_t beginTime = hrtMsprofSysCycleTime();
     647              :     // channelParam资源参数填充
     648            0 :     strncpy_s(channelParam.hcomId, HCOMID_MAX_LENGTH, commId.c_str(), HCOMID_MAX_LENGTH - 1);
     649            0 :     strncpy_s(channelParam.channelTag, TAG_MAX_LENGTH, tag.c_str(), TAG_MAX_LENGTH - 1);
     650            0 :     channelParam.engine = engine;
     651            0 :     channelParam.localUserRank = userRank_;
     652            0 :     channelParam.multiQpThreshold = GetExternalInputMultiQpThreshold();
     653              : 
     654              :     // 为device侧的channelList分配内存
     655            0 :     DeviceMem deviceChannelList = DeviceMem::alloc(listNum * sizeof(ChannelHandle));
     656            0 :     CHK_PTR_NULL(deviceChannelList.ptr());
     657            0 :     channelParam.channelList = static_cast<void*>(deviceChannelList.ptr());
     658            0 :     channelParam.listNum = listNum;
     659              : 
     660              :     // 将建链获取的远端数据填充到channelParam
     661            0 :     HcclResult ret = ParseChannelRemoteDataToMem(opTransportResponse, channelParam);
     662            0 :     if (ret != HCCL_SUCCESS) {
     663            0 :         HCCL_ERROR("[%s] ParseChannelRemoteDataToMem failed, return [%d].", __func__, ret);
     664            0 :         ReleaseChannelParam(channelParam);
     665            0 :         return ret;
     666              :     }
     667              : 
     668              :     // 创建局部流
     669            0 :     Stream localStream(StreamType::STREAM_TYPE_ONLINE);
     670            0 :     constexpr u32 aicpuStreamMode = 1;
     671            0 :     CHK_RET(hrtStreamSetMode(localStream.ptr(), aicpuStreamMode));
     672              : 
     673              :     // 将channelParam内部的host内存拷贝成device内存
     674            0 :     HcclIndOpChannelRemoteResV3 deviceChannelParam = channelParam;
     675            0 :     CHK_RET(DeepCopyH2DchannelParam(channelParam, deviceChannelParam));
     676              : 
     677            0 :     DeviceMem addr = DeviceMem::alloc(sizeof(deviceChannelParam));
     678            0 :     CHK_PTR_NULL(addr.ptr());
     679            0 :     CHK_RET(hrtMemSyncCopy(
     680              :         addr.ptr(), sizeof(deviceChannelParam), &deviceChannelParam, sizeof(deviceChannelParam),
     681              :         HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE));
     682              : 
     683              :     // 下kernel
     684            0 :     std::string kernelName = "RunAicpuIndOpChannelInit";
     685              :     struct InitTask {
     686              :         u64 context;
     687              :         bool isCustom;
     688              :     };
     689            0 :     InitTask customInitTask = {};
     690            0 :     customInitTask.context = reinterpret_cast<u64>(addr.ptr());
     691            0 :     customInitTask.isCustom = false;
     692              : 
     693            0 :     u16 timeOut = NOTIFY_DEFAULT_WAIT_TIME > std::numeric_limits<uint16_t>::max() ?
     694              :                       std::numeric_limits<uint16_t>::max() :
     695              :                       NOTIFY_DEFAULT_WAIT_TIME;
     696            0 :     CHK_RET(AicpuAclKernelLaunch(
     697              :         localStream.ptr(), reinterpret_cast<void*>(&customInitTask), sizeof(customInitTask), binHandle_, kernelName,
     698              :         true, timeOut));
     699            0 :     CHK_RET(hcclStreamSynchronize(localStream.ptr(), CommConfiger::GetInstance().GetCommConfigExecTimeOut(tag)));
     700              : 
     701              :     // 将device侧的channelList拷贝回host侧的channelList
     702            0 :     CHK_RET(hrtMemSyncCopy(
     703              :         channelList, listNum * sizeof(ChannelHandle), deviceChannelList.ptr(), listNum * sizeof(ChannelHandle),
     704              :         HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_HOST));
     705              : 
     706              :     // 手动释放channelParam中申请的内存
     707            0 :     CHK_RET(ReleaseChannelParam(channelParam));
     708            0 :     const std::string profName = "RunAicpuIndOpChannelInit";
     709            0 :     HCCL_DEBUG("[%s] RunAicpuIndOpChannelInit", __func__);
     710              :     // 上报初始化kernel的时间
     711            0 :     HcommProfilingReportKernel(beginTime, profName.c_str());
     712            0 :     return HCCL_SUCCESS;
     713            0 : }
     714              : 
     715              : const std::map<CommEngine, std::string> COMM_ENGINE_TYPE_STR_MAP{
     716              :     {CommEngine::COMM_ENGINE_CPU, "host_cpu"},     {CommEngine::COMM_ENGINE_CPU_TS, "host_cpu_ts"},
     717              :     {CommEngine::COMM_ENGINE_AICPU, "aicpu"},      {CommEngine::COMM_ENGINE_AICPU_TS, "aicpu_ts"},
     718              :     {CommEngine::COMM_ENGINE_AIV, "aiv"},          {CommEngine::COMM_ENGINE_CCU, "ccu"},
     719              :     {CommEngine::COMM_ENGINE_RESERVED, "reserved"}};
     720              : 
     721            0 : std::string GetCommEngineEnumStr(CommEngine engine)
     722              : {
     723            0 :     auto iter = COMM_ENGINE_TYPE_STR_MAP.find(engine);
     724            0 :     if (iter == COMM_ENGINE_TYPE_STR_MAP.end()) {
     725            0 :         return "CommEngine=" + std::to_string(engine);
     726              :     } else {
     727            0 :         return iter->second;
     728              :     }
     729              : }
     730              : 
     731            1 : HcclResult ChannelManager::ChannelCommCreate(
     732              :     const std::string& commId, CommEngine engine, const HcclChannelDesc* channelDescList, uint32_t listNum,
     733              :     ChannelHandle* channelList)
     734              : {
     735            1 :     CHK_RET(CheckChannelParam(engine, channelDescList, listNum));
     736              : 
     737              :     // channel复用,以tag + engine + remoterank + channelProtocol 作为channel标识
     738            0 :     std::vector<HcclChannelDesc> needCreateDescs;
     739            0 :     std::vector<uint32_t> needCreateIndices;
     740            0 :     std::string tag = commId;
     741            0 :     CHK_RET(PrepareHandleArray(tag, engine, channelDescList, listNum, channelList, needCreateDescs, needCreateIndices));
     742              : 
     743              :     // 对未复用的channelDesc进行建链
     744            0 :     if (needCreateDescs.size() > 0) {
     745              :         // 构造建链param
     746            0 :         OpCommTransport opCommTransport = BuildChannelRequests(needCreateDescs);
     747            0 :         std::string linkTag = commId + "_" + GetCommEngineEnumStr(engine);
     748            0 :         bool isAicpuModeEn = false;
     749            0 :         if (engine == COMM_ENGINE_AICPU || engine == COMM_ENGINE_AICPU_TS) {
     750            0 :             isAicpuModeEn = true;
     751              :         }
     752            0 :         std::vector<HcclMemHandle> memHandles = CollectMemHandles(needCreateDescs);
     753            0 :         CHK_RET(channelCallbacks_.indOpTransportAlloc(
     754              :             linkTag, opCommTransport, isAicpuModeEn, memHandles.data(), static_cast<uint32_t>(memHandles.size())));
     755              : 
     756            0 :         uint32_t level0 = 0;
     757            0 :         std::vector<LINK> links = opCommTransport[level0][level0].links;
     758            0 :         uint32_t newDescNum = needCreateDescs.size();
     759              :         // 创建host或device侧channel句柄
     760            0 :         if (isAicpuModeEn) {
     761              :             // Kernel下发恢复
     762            0 :             if (!callbacks_.getAicpuCommState()) {
     763            0 :                 HcclResult ret = callbacks_.kernelLaunchAicpuCommInit();
     764            0 :                 CHK_PRT_RET(
     765              :                     ret != HCCL_SUCCESS,
     766              :                     HCCL_ERROR("[%s] kernelLaunchAicpuCommInit failed, return [%d].", __func__, ret), ret);
     767            0 :                 callbacks_.setAicpuCommState(true);
     768              :             }
     769            0 :             std::unique_ptr<ChannelHandle[]> tmpChannelList = std::make_unique<ChannelHandle[]>(newDescNum);
     770            0 :             CHK_RET(AicpuChannelInit(commId, tag, engine, opCommTransport, tmpChannelList.get(), newDescNum));
     771            0 :             for (u32 i = 0; i < newDescNum; i++) {
     772            0 :                 uint32_t arrayIndex = needCreateIndices[i];
     773            0 :                 channelList[arrayIndex] = tmpChannelList[i];
     774            0 :                 CHK_RET(RegisterHandle(tag, engine, needCreateDescs[i], tmpChannelList[i]));
     775            0 :                 ChannelHandle channelHandle = reinterpret_cast<ChannelHandle>(links[i].get());
     776            0 :                 CHK_RET(RegisterHandleHDPair(tmpChannelList[i], channelHandle));
     777              :             }
     778            0 :         } else {
     779            0 :             for (u32 i = 0; i < newDescNum; i++) {
     780            0 :                 uint32_t arrayIndex = needCreateIndices[i];
     781            0 :                 ChannelHandle channelHandle = reinterpret_cast<ChannelHandle>(links[i].get());
     782            0 :                 channelList[arrayIndex] = channelHandle;
     783            0 :                 CHK_RET(RegisterHandle(tag, engine, needCreateDescs[i], channelHandle));
     784              :             }
     785              :         }
     786              :         // 保存link
     787            0 :         for (auto& link : links) {
     788              :             // 设置成员变量保存link
     789            0 :             channelLinks_.push_back(link);
     790              :         }
     791            0 :     }
     792            0 :     return HCCL_SUCCESS;
     793            0 : }
     794              : 
     795            0 : HcclResult ChannelManager::ChannelCommGetNotifyNum(ChannelHandle channel, uint32_t* notifyNum)
     796              : {
     797            0 :     CHK_RET(IsChannelExist(channel));
     798              :     ChannelHandle hostchannel;
     799            0 :     CHK_RET(GetHostChannel(channel, hostchannel));
     800              : 
     801            0 :     Transport* transportPtr = reinterpret_cast<Transport*>(hostchannel);
     802            0 :     *notifyNum = transportPtr->GetNotifyNum();
     803            0 :     return HCCL_SUCCESS;
     804              : }
     805              : 
     806            3 : HcclResult ChannelManager::ChannelCommDestroy(ChannelHandle* channelList, uint32_t channelNum)
     807              : {
     808            7 :     for (uint32_t i = 0; i < channelNum; ++i) {
     809            4 :         UnregisterHandle(channelList[i]);
     810            4 :         channelList[i] = 0;
     811              :     }
     812            3 :     return HCCL_SUCCESS;
     813              : }
     814              : 
     815            0 : HcclResult ChannelManager::ChannelCommGetHcclBuffer(ChannelHandle channel, CommBuffer* buffer)
     816              : {
     817              :     ChannelHandle hostchannel;
     818            0 :     CHK_RET(IsChannelExist(channel));
     819            0 :     CHK_RET(GetHostChannel(channel, hostchannel));
     820            0 :     Transport* transportPtr = reinterpret_cast<Transport*>(hostchannel);
     821              : 
     822            0 :     buffer->addr = nullptr;
     823            0 :     CHK_RET(transportPtr->GetRemoteMem(UserMemType::INPUT_MEM, &buffer->addr));
     824            0 :     CHK_PTR_NULL(buffer->addr);
     825            0 :     u64 tempSize = 0;
     826            0 :     CHK_RET(transportPtr->GetRemoteMemSize(UserMemType::INPUT_MEM, tempSize));
     827            0 :     buffer->size = static_cast<uint64_t>(tempSize);
     828            0 :     buffer->type = HCCL_MEM_TYPE_DEVICE;
     829            0 :     HCCL_INFO(
     830              :         "[%s]channel[%llu] channelKey[%s] get remote hccl buffer success, remote addr[%p], size[%llu]", __func__,
     831              :         channel, keyMap_[channel].c_str(), buffer->addr, buffer->size);
     832            0 :     return HCCL_SUCCESS;
     833              : }
     834              : 
     835            0 : HcclResult ChannelManager::ChannelCommGetRemoteMem(ChannelHandle channel, HcclMem** remoteMem, uint32_t* memNum)
     836              : {
     837            0 :     CHK_RET(IsChannelExist(channel));
     838              :     ChannelHandle hostchannel;
     839            0 :     CHK_RET(GetHostChannel(channel, hostchannel));
     840            0 :     Transport* transportPtr = reinterpret_cast<Transport*>(hostchannel);
     841              : 
     842            0 :     CHK_RET(transportPtr->GetIndOpRemoteMem(remoteMem, memNum));
     843            0 :     HCCL_INFO("[%s]get remote mem success, mem num[%u]", __func__, *memNum);
     844            0 :     return HCCL_SUCCESS;
     845              : }
     846              : 
     847          405 : HcclResult ChannelManager::ReleaseChannel()
     848              : {
     849          405 :     for (auto& link : channelLinks_) {
     850            0 :         if (link != nullptr) {
     851            0 :             if (link->DeInit() != HCCL_SUCCESS) {
     852            0 :                 HCCL_ERROR("[%s]transport[%p] deinit failed.", __func__, link.get());
     853              :             }
     854              :         }
     855              :     }
     856          405 :     channelLinks_.clear();
     857          405 :     return HCCL_SUCCESS;
     858              : }
     859              : 
     860          236 : HcclResult ChannelManager::SetHcclQos(u32 hcclQos)
     861              : {
     862          236 :     HCCL_INFO("[ChannelManager] [SetHcclQos] hcclQos[%u]", hcclQos);
     863          236 :     hcclQos_ = hcclQos;
     864          236 :     return HCCL_SUCCESS;
     865              : }
     866              : } // namespace hccl
        

Generated by: LCOV version 2.0-1