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