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