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              : #include "conn_local_cnt_notify_manager.h"
      11              : #include <set>
      12              : #include "communicator_impl.h"
      13              : #include "rdma_handle_manager.h"
      14              : 
      15              : namespace Hccl {
      16              : 
      17          514 : ConnLocalCntNotifyManager::ConnLocalCntNotifyManager(CommunicatorImpl *communicator) : comm(communicator)
      18              : {
      19          514 : }
      20              : 
      21          514 : ConnLocalCntNotifyManager::~ConnLocalCntNotifyManager()
      22              : {
      23          514 :     DECTOR_TRY_CATCH("ConnLocalCntNotifyManager", Destroy());
      24          514 : }
      25              : 
      26            4 : void ConnLocalCntNotifyManager::ApplyFor(u32 topicId, vector<LinkData> links)
      27              : {
      28           12 :     HCCL_INFO("in topicId=%u apply for now1", topicId);
      29              : 
      30            4 :     if (rtsNotifyPool.count(topicId) != 0) {
      31            3 :         HCCL_WARNING("Local notify for topicId[%u] already exists, no need to alloc.", topicId);
      32            1 :         return;
      33              :     }
      34              : 
      35            9 :     HCCL_INFO("topicId=%u apply for now", topicId);
      36              : 
      37              :     // 拿到并存储全部的portData
      38            3 :     set<std::pair<PortData, LinkProtocol>> ports;
      39            5 :     for (auto &link : links) {
      40            3 :         auto linkProtocol = link.GetLinkProtocol();
      41            3 :         bool ifUbProto = linkProtocol == LinkProtocol::UB_TP || linkProtocol == LinkProtocol::UB_CTP;
      42            3 :         if (link.GetType() != PortDeploymentType::DEV_NET || !ifUbProto) {
      43              :             string msg
      44            1 :                 = StringFormat("Unsupported %s of link %s", link.GetType().Describe().c_str(), link.Describe().c_str());
      45            1 :             THROW<InvalidParamsException>(msg);
      46            1 :         }
      47              : 
      48            6 :         HCCL_INFO("topicId=%u, linkData=%s", topicId, link.Describe().c_str());
      49              : 
      50            2 :         auto portData = link.GetLocalPort();
      51            2 :         ports.insert(std::make_pair(portData, linkProtocol));
      52              :     }
      53              : 
      54            2 :     u32 count = 2;
      55            2 :     rtsNotifyPool[topicId].resize(count);
      56              : 
      57            6 :     for (u32 i = 0; i < count; ++i) {
      58            4 :         rtsNotifyPool[topicId][i] = std::make_unique<RtsCntNotify>();
      59            8 :         for (auto &portLinkProtoPair : ports) {
      60            4 :             auto port = portLinkProtoPair.first;
      61            4 :             auto linkProtocol = portLinkProtoPair.second;
      62            4 :             RdmaHandle rdmaHandle = RdmaHandleManager::GetInstance().Get(comm->GetDevicePhyId(), port, linkProtocol);
      63            4 :             if (rdmaHandle == nullptr) {
      64              :                 string msg = StringFormat("Failed to get rdma handle for devicePhyId %u, port %u",
      65            0 :                                           comm->GetDevicePhyId(), port);
      66            0 :                 THROW<NullPtrException>(msg);
      67            0 :             }
      68            8 :             localCntNotifyPool[port][topicId].push_back(
      69            8 :                 std::make_unique<LocalCntNotify>(rdmaHandle, rtsNotifyPool[topicId][i].get()));
      70              :         }
      71              :     };
      72            3 : }
      73              : 
      74            3 : vector<RtsCntNotify *> ConnLocalCntNotifyManager::Get(u32 topicId)
      75              : {
      76            3 :     if (rtsNotifyPool.count(topicId) == 0) {
      77            3 :         HCCL_WARNING("Local notify for topicId[%u] not exists, no need to get.", topicId);
      78            1 :         return {};
      79              :     }
      80              : 
      81            2 :     vector<RtsCntNotify *> v;
      82            2 :     v.push_back(rtsNotifyPool[topicId][0].get());
      83            2 :     v.push_back(rtsNotifyPool[topicId][1].get());
      84            2 :     return v;
      85            2 : }
      86              : 
      87          514 : bool ConnLocalCntNotifyManager::Destroy()
      88              : {
      89          514 :     localCntNotifyPool.clear();
      90          514 :     rtsNotifyPool.clear();
      91          514 :     return true;
      92              : }
      93              : 
      94            5 : unordered_map<u32, vector<LocalCntNotify *>> ConnLocalCntNotifyManager::GetTopicIdCntNotifyMap(const PortData &portData)
      95              : {
      96           15 :     HCCL_INFO("GetTopicIdCntNotifyMap portData=%s", portData.Describe().c_str());
      97            5 :     unordered_map<u32, vector<LocalCntNotify *>> result;
      98            5 :     for (auto &it : localCntNotifyPool) {
      99            0 :         if (it.first != portData) {
     100            0 :             continue;
     101              :         }
     102            0 :         for (auto &topicIdIt : it.second) {
     103            0 :             for (auto &notify : topicIdIt.second) {
     104            0 :                 result[topicIdIt.first].push_back(notify.get());
     105              :             }
     106              :         }
     107              :     }
     108              : 
     109            5 :     return result;
     110            0 : }
     111              : 
     112              : } // namespace Hccl