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