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 "local_rma_buf_manager.h"
      12              : #include "internal_exception.h"
      13              : #include "rdma_handle_manager.h"
      14              : #include "local_ipc_rma_buffer_v2.h"
      15              : #include "local_rdma_rma_buffer_v2.h"
      16              : #include "local_ub_rma_buffer.h"
      17              : 
      18              : #include "log.h"
      19              : #include "stl_util.h"
      20              : #include "exception_util.h"
      21              : #include "communicator_impl.h"
      22              : 
      23              : namespace Hccl {
      24              : 
      25          332 : LocalRmaBufManager::LocalRmaBufManager(const CommunicatorImpl& communicator)
      26          332 :     : comm(const_cast<CommunicatorImpl*>(&communicator))
      27          332 : {}
      28              : 
      29          332 : LocalRmaBufManager::~LocalRmaBufManager() { DECTOR_TRY_CATCH("LocalRmaBufManager", Destroy()); }
      30              : 
      31           36 : bool LocalRmaBufManager::IsExist(const string& opTag, const PortData& portData, BufferType bufferType)
      32              : {
      33           53 :     return bufs.find(opTag) != bufs.end() && bufs[opTag].find(portData) != bufs[opTag].end()
      34           53 :            && bufs[opTag][portData].find(bufferType) != bufs[opTag][portData].end();
      35              : }
      36              : 
      37           14 : LocalRmaBuffer* LocalRmaBufManager::Reg(
      38              :     const string& opTag, BufferType bufferType, std::shared_ptr<Buffer> buffer, const PortData& portData,
      39              :     LinkProtocol linkProtocol)
      40              : {
      41           14 :     if (buffer == nullptr) {
      42            0 :         HCCL_ERROR("input buffer is null");
      43            0 :         return nullptr;
      44              :     }
      45           42 :     HCCL_INFO(
      46              :         "LocalRmaBufManager::Reg, buffer[%s], opTag[%s], bufferType[%u], portData[%s]", buffer->Describe().c_str(),
      47              :         opTag.c_str(), bufferType, portData.Describe().c_str());
      48           14 :     if (IsExist(opTag, portData, bufferType)) {
      49              :         string msg = StringFormat(
      50              :             "opTag=%s bufferType=%s, buffer=%s already reg to portData=%s", opTag.c_str(),
      51            1 :             bufferType.Describe().c_str(), buffer->Describe().c_str(), portData.Describe().c_str());
      52            3 :         HCCL_DEBUG(msg.c_str());
      53            1 :         return bufs[opTag][portData][bufferType].get();
      54            1 :     }
      55           13 :     if (portData.GetType() == PortDeploymentType::P2P) {
      56            0 :         bufs[opTag][portData][bufferType] = make_unique<LocalIpcRmaBuffer>(buffer);
      57            0 :         return bufs[opTag][portData][bufferType].get();
      58              :     } else {
      59           13 :         if (portData.GetProto() == LinkProtoType::RDMA) {
      60              :             RdmaHandle rdmaHandle
      61            0 :                 = RdmaHandleManager::GetInstance().Get(comm->GetDevicePhyId(), portData, linkProtocol);
      62            0 :             bufs[opTag][portData][bufferType] = make_unique<LocalRdmaRmaBuffer>(buffer, rdmaHandle);
      63            0 :             return bufs[opTag][portData][bufferType].get();
      64           13 :         } else if (portData.GetProto() == LinkProtoType::UB) {
      65           36 :             HCCL_INFO(
      66              :                 "LocalRmaBufManager::Reg, comm->GetOpAiCpuTSFeatureFlag[%d], comm->GetOpAivFeatureFlag[%d]",
      67              :                 comm->GetOpAiCpuTSFeatureFlag(), comm->GetOpAivFeatureFlag());
      68           12 :             if (comm->GetOpAiCpuTSFeatureFlag() || comm->GetOpAivFeatureFlag()) { // 算子粒度
      69            4 :                 bufs[opTag][portData][bufferType] = make_unique<LocalUbRmaBuffer>(buffer);
      70              :             } else {
      71              :                 RdmaHandle rdmaHandle
      72            8 :                     = RdmaHandleManager::GetInstance().Get(comm->GetDevicePhyId(), portData, linkProtocol);
      73            8 :                 bufs[opTag][portData][bufferType] = make_unique<LocalUbRmaBuffer>(buffer, rdmaHandle);
      74              :             }
      75           12 :             return bufs[opTag][portData][bufferType].get();
      76              :         }
      77              :         // 待修改: 仅支持 P2P 和 RDMA
      78            1 :         string msg = StringFormat("PortData=%s is error", portData.Describe().c_str());
      79            4 :         MACRO_THROW(InternalException, msg);
      80            1 :     }
      81              : }
      82              : 
      83           22 : LocalRmaBuffer* LocalRmaBufManager::Get(const string& opTag, const PortData& portData, BufferType bufferType)
      84              : {
      85           22 :     if (IsExist(opTag, portData, bufferType)) { // if localRmaBuffer exists
      86           24 :         HCCL_INFO(
      87              :             "[LocalRmaBufManager][%s] LocalUbRmaBuffer[%s]", __func__,
      88              :             bufs[opTag][portData][bufferType]->Describe().c_str());
      89            8 :         return bufs[opTag][portData][bufferType].get();
      90              :     }
      91           42 :     HCCL_WARNING(
      92              :         "LocalRmaBuffer doesn't exist:opTag[%s], bufferType[%u], portData[%s]", opTag.c_str(), bufferType,
      93              :         portData.Describe().c_str());
      94           14 :     return nullptr;
      95              : }
      96              : 
      97          335 : void LocalRmaBufManager::Destroy() { bufs.clear(); }
      98              : 
      99            0 : LocalRmaBuffer* LocalRmaBufManager::Get(const PortData& portData)
     100              : {
     101            0 :     if (Contain(ccuBufs, portData)) {
     102            0 :         return ccuBufs[portData].get();
     103              :     }
     104            0 :     HCCL_WARNING("LocalRmaBuffer doesn't exist at port[%s].", portData.Describe().c_str());
     105            0 :     return nullptr;
     106              : }
     107              : 
     108            1 : HcclResult LocalRmaBufManager::Dereg(const string& opTag)
     109              : {
     110            1 :     if (bufs.find(opTag) == bufs.end()) {
     111            3 :         HCCL_WARNING("[LocalRmaBufManager::%s] opTag[%s] Cannot find Buffer in bufs.", __func__, opTag.c_str());
     112            1 :         return HCCL_SUCCESS;
     113              :     }
     114            0 :     bufs.erase(opTag);
     115            0 :     return HCCL_SUCCESS;
     116              : }
     117              : 
     118              : } // namespace Hccl