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