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 "urma_endpoint.h"
11 : #include <algorithm>
12 : #include "endpoint_mgr.h"
13 : #include "log.h"
14 : #include "hccl/hccl_res.h"
15 : #include "urma_mem.h"
16 : #include "proc_reged_mem_mgr_cache.h"
17 : #include "adapter_rts_common.h"
18 : #include "server_socket_manager.h"
19 : #include "ra_rs_comm.h"
20 :
21 : namespace hcomm {
22 :
23 34 : UrmaEndpoint::UrmaEndpoint(const EndpointDesc& endpointDesc) : Endpoint(endpointDesc) {}
24 :
25 68 : UrmaEndpoint::~UrmaEndpoint() noexcept
26 : {
27 34 : std::lock_guard<std::mutex> lock(portMutex_);
28 34 : if (dynamicPort_ != HCCL_INVALID_PORT) {
29 2 : ServerSocketStopListenImpl(dynamicPort_);
30 : }
31 34 : dynamicPort_ = HCCL_INVALID_PORT;
32 34 : ProcRegedMemMgrCache::GetInstance().Release(cacheKey_);
33 68 : }
34 :
35 18 : HcclResult UrmaEndpoint::Init()
36 : {
37 18 : HCCL_INFO("[%s] localEndpoint protocol[%d]", __func__, endpointDesc_.protocol);
38 :
39 18 : if (endpointDesc_.loc.locType != ENDPOINT_LOC_TYPE_DEVICE) {
40 0 : HCCL_ERROR(
41 : "[UrmaEndpoint][%s] endpointDesc.loc.locType[%d] only support ENDPOINT_LOC_TYPE_DEVICE", __func__,
42 : endpointDesc_.loc.locType);
43 0 : return HCCL_E_PARA;
44 : }
45 :
46 18 : Hccl::IpAddress ipAddr{};
47 18 : HcclResult ret = CommAddrToIpAddress(endpointDesc_.commAddr, ipAddr);
48 18 : if (ret != HCCL_SUCCESS) {
49 0 : HCCL_ERROR("call CommAddrToIpAddress failed");
50 0 : return ret;
51 : }
52 :
53 : u32 devPhyId;
54 : s32 deviceLogicId;
55 18 : ret = hrtGetDevice(&deviceLogicId);
56 18 : if (ret != HCCL_SUCCESS) {
57 0 : HCCL_ERROR("call hrtGetDevice failed, deviceLogicId[%d]", deviceLogicId);
58 0 : return ret;
59 : }
60 18 : Hccl::HccpHdcManager::GetInstance().Init(deviceLogicId);
61 18 : ret = hrtGetDevicePhyIdByIndex(static_cast<uint32_t>(deviceLogicId), devPhyId);
62 18 : if (ret != HCCL_SUCCESS) {
63 0 : HCCL_ERROR("call hrtGetDevicePhyIdByIndex failed, deviceLogicId[%d], devPhyId[%u]", deviceLogicId, devPhyId);
64 0 : return ret;
65 : }
66 :
67 18 : if (endpointDesc_.loc.device.devPhyId != devPhyId) {
68 2 : HCCL_WARNING(
69 : "[UrmaEndpoint][%s] endpointDesc.loc.device.devPhyId[%u] incorrect", __func__,
70 : endpointDesc_.loc.device.devPhyId);
71 : endpointDesc_.loc.device.devPhyId
72 2 : = devPhyId; // 当前endpointDesc.loc.device.devPhyId不准，暂时由查询的devPhyId赋值
73 : }
74 :
75 18 : auto& rdmaHandleMgr = Hccl::RdmaHandleManager::GetInstance();
76 18 : EXCEPTION_CATCH(
77 : ctxHandle_ = static_cast<void*>(rdmaHandleMgr.GetByIp(endpointDesc_.loc.device.devPhyId, ipAddr)),
78 : return HCCL_E_PARA);
79 17 : CHK_PTR_NULL(ctxHandle_);
80 17 : HCCL_INFO(
81 : "%s success, devPhyId[%u], ipAddr[%s], ctxHandle[%p]", __func__, devPhyId, ipAddr.Describe().c_str(),
82 : ctxHandle_);
83 :
84 17 : cacheKey_ = MemMgrCacheKey{devPhyId, COMM_PROTOCOL_UBC_CTP, ipAddr, LocTypeToPortType(endpointDesc_.loc.locType)};
85 17 : auto& cache = ProcRegedMemMgrCache::GetInstance();
86 34 : EXCEPTION_CATCH(
87 : this->regedMemMgr_ = cache.GetOrCreate(
88 : cacheKey_,
89 : [this]() {
90 : auto m = std::make_shared<UbRegedMemMgr>();
91 : m->rdmaHandle_ = this->ctxHandle_;
92 : return m;
93 : }),
94 : return HCCL_E_INTERNAL);
95 :
96 : // ccu模式专用的资源分配器
97 17 : ccuChannelCtxPool_.reset(new (std::nothrow) CcuChannelCtxPool(deviceLogicId));
98 17 : CHK_PTR_NULL(ccuChannelCtxPool_);
99 :
100 17 : return HcclResult::HCCL_SUCCESS;
101 : }
102 :
103 3 : HcclResult UrmaEndpoint::ServerSocketListen(const uint32_t port)
104 : {
105 3 : if (endpointDesc_.loc.locType != ENDPOINT_LOC_TYPE_DEVICE) {
106 1 : HCCL_INFO(
107 : "[UrmaEndpoint][%s] endpointDesc.loc.locType[%d] skip create ServerSocket", __func__,
108 : endpointDesc_.loc.locType);
109 1 : return HCCL_SUCCESS;
110 : }
111 :
112 2 : Hccl::IpAddress ipaddr{};
113 2 : CHK_RET(CommAddrToIpAddress(endpointDesc_.commAddr, ipaddr));
114 :
115 2 : HCCL_INFO("[UrmaEndpoint][%s] endpointDesc_.loc.device.devPhyId %u", __func__, endpointDesc_.loc.device.devPhyId);
116 :
117 2 : Hccl::DevNetPortType type = Hccl::DevNetPortType(Hccl::ConnectProtoType::UB);
118 : Hccl::PortData localPort
119 2 : = Hccl::PortData(static_cast<Hccl::RankId>(endpointDesc_.loc.device.devPhyId), type, 0, ipaddr);
120 :
121 2 : HCCL_INFO(
122 : "[UrmaEndpoint][%s] devicePhyId[%u] localPort[%s]", __func__, endpointDesc_.loc.device.devPhyId,
123 : localPort.Describe().c_str());
124 2 : uint32_t requestPort = port;
125 2 : CHK_RET(ServerSocketManager::GetInstance().ServerSocketStartListen(
126 : localPort, Hccl::NicType::DEVICE_NIC_TYPE, endpointDesc_.loc.device.devPhyId, &requestPort));
127 1 : return HCCL_SUCCESS;
128 : }
129 :
130 2 : inline HcclResult UrmaEndpoint::ServerSocketStopListenImpl(const uint32_t port)
131 : {
132 2 : Hccl::IpAddress ipAddr{};
133 2 : CHK_RET(CommAddrToIpAddress(endpointDesc_.commAddr, ipAddr));
134 2 : Hccl::DevNetPortType type = Hccl::DevNetPortType(Hccl::ConnectProtoType::UB);
135 : Hccl::PortData localPort
136 2 : = Hccl::PortData(static_cast<Hccl::RankId>(endpointDesc_.loc.device.devPhyId), type, 0, ipAddr);
137 :
138 2 : CHK_RET(ServerSocketManager::GetInstance().ServerSocketStopListen(localPort, Hccl::NicType::DEVICE_NIC_TYPE, port));
139 0 : return HCCL_SUCCESS;
140 : }
141 :
142 0 : HcclResult UrmaEndpoint::ServerSocketStopListen(const uint32_t port) { return ServerSocketStopListenImpl(port); }
143 :
144 6 : HcclResult UrmaEndpoint::ServerSocketGetListenPort(uint32_t* port)
145 : {
146 6 : std::lock_guard<std::mutex> lock(portMutex_);
147 6 : if (endpointDesc_.loc.locType != ENDPOINT_LOC_TYPE_DEVICE) {
148 1 : HCCL_INFO(
149 : "[UrmaEndpoint][%s] endpointDesc.loc.locType[%d] skip create ServerSocket", __func__,
150 : endpointDesc_.loc.locType);
151 1 : return HCCL_SUCCESS;
152 : }
153 :
154 5 : HCCL_INFO("[UrmaEndpoint][%s] endpointDesc_.loc.device.devPhyId %u", __func__, endpointDesc_.loc.device.devPhyId);
155 5 : Hccl::IpAddress ipaddr{};
156 5 : CHK_RET(CommAddrToIpAddress(endpointDesc_.commAddr, ipaddr));
157 :
158 5 : Hccl::DevNetPortType type = Hccl::DevNetPortType(Hccl::ConnectProtoType::UB);
159 : Hccl::PortData localPort
160 5 : = Hccl::PortData(static_cast<Hccl::RankId>(endpointDesc_.loc.device.devPhyId), type, 0, ipaddr);
161 :
162 5 : HCCL_INFO(
163 : "[UrmaEndpoint][%s] devicePhyId[%u] localPort[%s]", __func__, endpointDesc_.loc.device.devPhyId,
164 : localPort.Describe().c_str());
165 :
166 : // 已有监听端口则直接返回
167 5 : if (dynamicPort_ != HCCL_INVALID_PORT) {
168 1 : *port = dynamicPort_;
169 1 : HCCL_INFO("[UrmaEndpoint::%s] already listening, return existing port[%u]", __func__, dynamicPort_);
170 1 : return HCCL_SUCCESS;
171 : }
172 4 : uint32_t requestPort = 0;
173 4 : CHK_RET(ServerSocketManager::GetInstance().ServerSocketStartListen(
174 : localPort, Hccl::NicType::DEVICE_NIC_TYPE, endpointDesc_.loc.device.devPhyId, &requestPort));
175 3 : if (requestPort == 0 || requestPort == HCCL_INVALID_PORT) {
176 2 : HCCL_ERROR("[UrmaEndpoint::%s] get listen port failed, port is invalid.", __func__);
177 2 : return HCCL_E_NETWORK;
178 : }
179 1 : dynamicPort_ = requestPort;
180 1 : *port = dynamicPort_;
181 1 : return HCCL_SUCCESS;
182 6 : }
183 :
184 0 : HcclResult UrmaEndpoint::RegisterMemory(HcommMem mem, const char* memTag, void** memHandle)
185 : {
186 0 : CHK_RET(this->regedMemMgr_->RegisterMemory(mem, memTag, memHandle));
187 0 : return HCCL_SUCCESS;
188 : }
189 :
190 0 : HcclResult UrmaEndpoint::UnregisterMemory(void* memHandle)
191 : {
192 0 : CHK_RET(this->regedMemMgr_->UnregisterMemory(memHandle));
193 0 : return HCCL_SUCCESS;
194 : }
195 :
196 0 : HcclResult UrmaEndpoint::MemoryExport(void* memHandle, void** memDesc, uint32_t* memDescLen)
197 : {
198 0 : CHK_RET(this->regedMemMgr_->MemoryExport(this->endpointDesc_, memHandle, memDesc, memDescLen));
199 0 : return HCCL_SUCCESS;
200 : }
201 :
202 0 : HcclResult UrmaEndpoint::MemoryImport(const void* memDesc, uint32_t descLen, HcommMem* outMem)
203 : {
204 0 : CHK_RET(this->regedMemMgr_->MemoryImport(memDesc, descLen, outMem));
205 0 : return HCCL_SUCCESS;
206 : }
207 :
208 0 : HcclResult UrmaEndpoint::MemoryUnimport(const void* memDesc, uint32_t descLen)
209 : {
210 0 : CHK_RET(this->regedMemMgr_->MemoryUnimport(memDesc, descLen));
211 0 : return HCCL_SUCCESS;
212 : }
213 :
214 0 : HcclResult UrmaEndpoint::GetAllMemHandles(void** memHandles, uint32_t* memHandleNum)
215 : {
216 0 : CHK_RET(this->regedMemMgr_->GetAllMemHandles(memHandles, memHandleNum));
217 0 : return HCCL_SUCCESS;
218 : }
219 :
220 15 : CcuChannelCtxPool* UrmaEndpoint::GetCcuChannelCtxPool() { return ccuChannelCtxPool_.get(); }
221 :
222 4 : HcclResult UrmaEndpoint::GetAsyncEvents(uint32_t devPhyId, struct AsyncEvent events[], uint32_t& num)
223 : {
224 4 : uint32_t interfaceVersion{0};
225 :
226 : int ret;
227 : // 对RaCtxGetAsyncEvents接口的版本检验
228 4 : ret = RaGetInterfaceVersion(devPhyId, RA_RS_CTX_GET_ASYNC_EVENTS, &interfaceVersion);
229 4 : if (ret != 0) {
230 0 : HCCL_ERROR(
231 : "[%s] devPhyId[%u] epHandle[%p] RaGetInterfaceVersion failed, ret[%d]", __func__, devPhyId, this, ret);
232 0 : return HCCL_E_INTERNAL;
233 : }
234 :
235 4 : if (interfaceVersion <= 1) {
236 1 : HCCL_ERROR(
237 : "[%s] devPhyId[%u] epHandle[%p] version[%u] not support", __func__, devPhyId, this, interfaceVersion);
238 1 : return HCCL_E_NOT_SUPPORT;
239 : }
240 :
241 3 : if (!IsCtxHandleValid()) {
242 1 : HCCL_ERROR(
243 : "[%s] devPhyId[%u] ctxHandle_[%p] has been invalidated, "
244 : "RdmaHandleManager may have DeInit this device",
245 : __func__, devPhyId, ctxHandle_);
246 1 : return HCCL_E_INTERNAL;
247 : }
248 :
249 2 : ret = RaCtxGetAsyncEvents(ctxHandle_, events, &num);
250 2 : if (ret != 0) {
251 1 : HCCL_ERROR(
252 : "[%s] devPhyId[%u] epHandle[%p] RaCtxGetAsyncEvents failed, ctxHandle[%p] ret[%d]", __func__, devPhyId,
253 : this, (void*)ctxHandle_, ret);
254 1 : return HCCL_E_INTERNAL;
255 : }
256 1 : return HCCL_SUCCESS;
257 : }
258 :
259 : } // namespace hcomm
|
