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