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 "endpoint_mgr.h"
12 : #include "hccl_mem_defs.h"
13 : #include "cpu_roce_endpoint.h"
14 : #include "hccl/hccl_res.h"
15 : #include "log.h"
16 : #include "roce_mem.h"
17 : #include "proc_reged_mem_mgr_cache.h"
18 : #include "host_socket_handle_manager.h"
19 : #include "adapter_rts_common.h"
20 : #include "hccp_peer_manager.h"
21 : #include "server_socket_manager.h"
22 : #include "hccp.h"
23 :
24 : using Hccl::HcclException;
25 : using std::exception;
26 : using std::string;
27 :
28 : namespace hcomm {
29 132 : CpuRoceEndpoint::CpuRoceEndpoint(const EndpointDesc& endpointDesc) : Endpoint(endpointDesc) {}
30 :
31 262 : CpuRoceEndpoint::~CpuRoceEndpoint() noexcept
32 : {
33 132 : std::lock_guard<std::mutex> lock(portMutex_);
34 132 : if (dynamicPort_ != HCCL_INVALID_PORT) {
35 0 : ServerSocketStopListenImpl(dynamicPort_);
36 : }
37 132 : dynamicPort_ = HCCL_INVALID_PORT;
38 132 : ProcRegedMemMgrCache::GetInstance().Release(cacheKey_);
39 262 : }
40 :
41 132 : HcclResult CpuRoceEndpoint::Init()
42 : {
43 132 : HCCL_INFO("[%s] localEndpoint protocol[%d]", __func__, endpointDesc_.protocol);
44 :
45 132 : if (endpointDesc_.loc.locType != ENDPOINT_LOC_TYPE_HOST) {
46 0 : HCCL_INFO("[CpuRoceEndpoint][%s] CpuRoceEndpoint not support device", __func__);
47 0 : return HCCL_E_NOT_SUPPORT;
48 : }
49 132 : Hccl::IpAddress ipAddr{};
50 132 : CHK_RET(CommAddrToIpAddress(endpointDesc_.commAddr, ipAddr));
51 132 : s32 devId = 0;
52 132 : CHK_RET(hrtGetDevice(&devId));
53 132 : EXCEPTION_CATCH(Hccl::HccpPeerManager::GetInstance().Init(devId), return HCCL_E_INTERNAL);
54 132 : u32 devPhyId = 0;
55 132 : CHK_RET(hrtGetDevicePhyIdByIndex(devId, devPhyId));
56 132 : auto& rdmaHandleMgr = Hccl::RdmaHandleManager::GetInstance();
57 132 : TRY_CATCH_RETURN(
58 : ctxHandle_ = static_cast<void*>(
59 : rdmaHandleMgr.GetByAddr(devPhyId, Hccl::LinkProtoType::RDMA, ipAddr, Hccl::PortDeploymentType::HOST_NET)));
60 132 : CHK_PTR_NULL(ctxHandle_);
61 131 : HCCL_INFO(
62 : "CpuRoceEndpoint::%s success, devPhyId[%u], ipAddr[%s], ctxHandle[%p]", __func__, devPhyId,
63 : ipAddr.Describe().c_str(), ctxHandle_);
64 :
65 131 : cacheKey_ = MemMgrCacheKey{devPhyId, COMM_PROTOCOL_ROCE, ipAddr, LocTypeToPortType(endpointDesc_.loc.locType)};
66 131 : auto& cache = ProcRegedMemMgrCache::GetInstance();
67 145 : EXCEPTION_CATCH(
68 : regedMemMgr_ = cache.GetOrCreate(
69 : cacheKey_,
70 : [this]() {
71 : auto m = std::make_shared<RoceRegedMemMgr>();
72 : m->rdmaHandle_ = this->ctxHandle_;
73 : return m;
74 : }),
75 : return HCCL_E_PARA);
76 131 : return HCCL_SUCCESS;
77 : }
78 :
79 4 : HcclResult CpuRoceEndpoint::ServerSocketListen(const uint32_t port)
80 : {
81 4 : Hccl::IpAddress ipAddr{};
82 4 : CHK_RET(CommAddrToIpAddress(endpointDesc_.commAddr, ipAddr));
83 :
84 4 : s32 devId = 0;
85 4 : CHK_RET(hrtGetDevice(&devId));
86 4 : u32 devPhyId = 0;
87 4 : CHK_RET(hrtGetDevicePhyIdByIndex(devId, devPhyId));
88 :
89 4 : Hccl::DevNetPortType type = Hccl::DevNetPortType(Hccl::ConnectProtoType::RDMA);
90 4 : Hccl::PortData localPort = Hccl::PortData(devPhyId, type, 0, ipAddr);
91 :
92 4 : HCCL_INFO("[CpuRoceEndpoint::%s] devicePhyId[%u] ipAddress[%s]", __func__, devPhyId, ipAddr.Describe().c_str());
93 :
94 4 : uint32_t requestPort = port;
95 4 : CHK_RET(ServerSocketManager::GetInstance().ServerSocketStartListen(
96 : localPort, Hccl::NicType::HOST_NIC_TYPE, devPhyId, &requestPort));
97 4 : return HCCL_SUCCESS;
98 : }
99 :
100 2 : inline HcclResult CpuRoceEndpoint::ServerSocketStopListenImpl(const uint32_t port)
101 : {
102 2 : Hccl::IpAddress ipAddr{};
103 2 : CHK_RET(CommAddrToIpAddress(endpointDesc_.commAddr, ipAddr));
104 :
105 2 : s32 devId = 0;
106 2 : CHK_RET(hrtGetDevice(&devId));
107 2 : u32 devPhyId = 0;
108 2 : CHK_RET(hrtGetDevicePhyIdByIndex(devId, devPhyId));
109 :
110 2 : Hccl::DevNetPortType type = Hccl::DevNetPortType(Hccl::ConnectProtoType::RDMA);
111 2 : Hccl::PortData localPort = Hccl::PortData(devPhyId, type, 0, ipAddr);
112 2 : CHK_RET(ServerSocketManager::GetInstance().ServerSocketStopListen(localPort, Hccl::NicType::HOST_NIC_TYPE, port));
113 :
114 2 : return HCCL_SUCCESS;
115 : }
116 :
117 2 : HcclResult CpuRoceEndpoint::ServerSocketStopListen(const uint32_t port) { return ServerSocketStopListenImpl(port); }
118 :
119 0 : HcclResult CpuRoceEndpoint::ServerSocketGetListenPort(uint32_t* port)
120 : {
121 0 : std::lock_guard<std::mutex> lock(portMutex_);
122 0 : CHK_PTR_NULL(port);
123 0 : s32 devId = 0;
124 0 : CHK_RET(hrtGetDevice(&devId));
125 0 : u32 devPhyId = 0;
126 0 : CHK_RET(hrtGetDevicePhyIdByIndex(devId, devPhyId));
127 :
128 0 : Hccl::IpAddress ipAddr{};
129 0 : CHK_RET(CommAddrToIpAddress(endpointDesc_.commAddr, ipAddr));
130 :
131 0 : Hccl::DevNetPortType type = Hccl::DevNetPortType(Hccl::ConnectProtoType::RDMA);
132 0 : Hccl::PortData localPort = Hccl::PortData(devPhyId, type, 0, ipAddr);
133 :
134 0 : HCCL_INFO("[CpuRoceEndpoint::%s] devicePhyId[%u] ipAddress[%s]", __func__, devPhyId, ipAddr.Describe().c_str());
135 :
136 : // 已有监听端口则直接返回
137 0 : if (dynamicPort_ != HCCL_INVALID_PORT) {
138 0 : *port = dynamicPort_;
139 0 : HCCL_INFO("[CpuRoceEndpoint::%s] already listening, return existing port[%u]", __func__, dynamicPort_);
140 0 : return HCCL_SUCCESS;
141 : }
142 0 : uint32_t requestPort = 0;
143 0 : CHK_RET(ServerSocketManager::GetInstance().ServerSocketStartListen(
144 : localPort, Hccl::NicType::HOST_NIC_TYPE, devPhyId, &requestPort));
145 0 : if (requestPort == 0 || requestPort == HCCL_INVALID_PORT) {
146 0 : HCCL_ERROR("[CpuRoceEndpoint::%s] get listen port failed, port is invalid", __func__);
147 0 : return HCCL_E_NETWORK;
148 : }
149 0 : dynamicPort_ = requestPort;
150 0 : *port = dynamicPort_;
151 0 : return HCCL_SUCCESS;
152 0 : }
153 :
154 1 : HcclResult CpuRoceEndpoint::RegisterMemory(HcommMem mem, const char* memTag, void** memHandle)
155 : {
156 1 : CHK_RET(this->regedMemMgr_->RegisterMemory(mem, memTag, memHandle));
157 0 : return HCCL_SUCCESS;
158 : }
159 :
160 2 : HcclResult CpuRoceEndpoint::UnregisterMemory(void* memHandle)
161 : {
162 2 : CHK_RET(this->regedMemMgr_->UnregisterMemory(memHandle));
163 0 : return HCCL_SUCCESS;
164 : }
165 :
166 0 : HcclResult CpuRoceEndpoint::MemoryExport(void* memHandle, void** memDesc, uint32_t* memDescLen)
167 : {
168 0 : CHK_RET(this->regedMemMgr_->MemoryExport(this->endpointDesc_, memHandle, memDesc, memDescLen));
169 0 : return HCCL_SUCCESS;
170 : }
171 :
172 0 : HcclResult CpuRoceEndpoint::MemoryImport(const void* memDesc, uint32_t descLen, HcommMem* outMem)
173 : {
174 0 : CHK_RET(this->regedMemMgr_->MemoryImport(memDesc, descLen, outMem));
175 0 : return HCCL_SUCCESS;
176 : }
177 :
178 0 : HcclResult CpuRoceEndpoint::MemoryUnimport(const void* memDesc, uint32_t descLen)
179 : {
180 0 : CHK_RET(this->regedMemMgr_->MemoryUnimport(memDesc, descLen));
181 0 : return HCCL_SUCCESS;
182 : }
183 :
184 0 : HcclResult CpuRoceEndpoint::GetAllMemHandles(void** memHandles, uint32_t* memHandleNum)
185 : {
186 0 : CHK_RET(this->regedMemMgr_->GetAllMemHandles(memHandles, memHandleNum));
187 0 : return HCCL_SUCCESS;
188 : }
189 :
190 65 : HcclResult CpuRoceEndpoint::GetCapabilities(Capabilities& caps)
191 : {
192 65 : HCCL_INFO("[CpuRoceEndpoint::%s] START.", __func__);
193 : static constexpr uint64_t RDMA_MAX_WR_LENGTH = 1ULL * 1024 * 1024 * 1024; // 单次RDMA操作最大长度1GB
194 65 : if (!isCapabilitiesAvailable_) {
195 : // 待 HCCP 提供查询设备支持的最大发送消息的接口后，查询设备实际值。
196 64 : capabilities_.maxMsgSize = RDMA_MAX_WR_LENGTH;
197 64 : uint32_t ret = RaGetLbMax(this->regedMemMgr_->rdmaHandle_, &(capabilities_.lbMax));
198 64 : HCCL_DEBUG("[CpuRoceEndpoint::GetCapabilities] lbMax = %d.", capabilities_.lbMax);
199 64 : CHK_PRT_RET(
200 : ret != 0,
201 : HCCL_ERROR(
202 : "[CpuRoceEndpoint::GetCapabilities][GetLbMax]errNo[0x%016llx] RaGetLbMax fail. "
203 : "return[%u], params: rdmaHandle[%p], lbMax[%d]",
204 : HCCL_ERROR_CODE(HCCL_E_NETWORK), ret, this->regedMemMgr_->rdmaHandle_, capabilities_.lbMax),
205 : HCCL_E_NETWORK);
206 64 : isCapabilitiesAvailable_ = true;
207 : }
208 65 : caps = capabilities_;
209 65 : HCCL_INFO("[CpuRoceEndpoint::%s] SUCCESS.", __func__);
210 65 : return HCCL_SUCCESS;
211 : }
212 : } // namespace hcomm
|
