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
|