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