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 "cpu_urma_endpoint.h"
11 : #include <algorithm>
12 : #include "endpoint_mgr.h"
13 : #include "log.h"
14 : #include "urma_mem.h"
15 : #include "proc_reged_mem_mgr_cache.h"
16 : #include "adapter_rts_common.h"
17 : #include "server_socket_manager.h"
18 : #include "hccp_peer_manager.h"
19 :
20 : namespace hcomm {
21 23 : CpuUrmaEndpoint::CpuUrmaEndpoint(const EndpointDesc &endpointDesc)
22 23 : : Endpoint(endpointDesc)
23 : {
24 23 : }
25 :
26 46 : CpuUrmaEndpoint::~CpuUrmaEndpoint() noexcept
27 : {
28 23 : std::lock_guard<std::mutex> lock(portMutex_);
29 23 : if (dynamicPort_ != HCCL_INVALID_PORT) {
30 1 : ServerSocketStopListenImpl(dynamicPort_);
31 : }
32 23 : dynamicPort_ = HCCL_INVALID_PORT;
33 23 : ProcRegedMemMgrCache::GetInstance().Release(cacheKey_);
34 46 : }
35 :
36 23 : HcclResult CpuUrmaEndpoint::Init()
37 : {
38 23 : HCCL_INFO("[%s] localEndpoint protocol[%d]", __func__, endpointDesc_.protocol);
39 :
40 23 : Hccl::IpAddress ipAddr{};
41 23 : CHK_RET(CommAddrToIpAddress(endpointDesc_.commAddr, ipAddr));
42 : u32 devPhyId;
43 : s32 deviceLogicId;
44 23 : HcclResult ret = hrtGetDevice(&deviceLogicId);
45 23 : if(ret != HCCL_SUCCESS){
46 0 : HCCL_ERROR("call hrtGetDevice failed, deviceLogicId[%d]", deviceLogicId);
47 0 : return ret;
48 : }
49 23 : RaSocketSetWhiteListStatus(1); // PEER模式需要手动开启白名单模式
50 23 : Hccl::HccpPeerManager::GetInstance().Init(deviceLogicId);
51 23 : ret = hrtGetDevicePhyIdByIndex(static_cast<uint32_t>(deviceLogicId), devPhyId);
52 23 : if(ret != HCCL_SUCCESS){
53 0 : HCCL_ERROR("call hrtGetDevicePhyIdByIndex failed, deviceLogicId[%d], devPhyId[%u]", deviceLogicId, devPhyId);
54 0 : return ret;
55 : }
56 23 : auto &rdmaHandleMgr = Hccl::RdmaHandleManager::GetInstance();
57 23 : ctxHandle_ = static_cast<void *>(
58 23 : rdmaHandleMgr.GetByAddr(devPhyId, Hccl::LinkProtoType::UB, ipAddr, Hccl::PortDeploymentType::HOST_NET));
59 23 : CHK_PTR_NULL(ctxHandle_);
60 23 : HCCL_INFO("CpuUrmaEndpoint::%s success, devPhyId[%u], ipAddr[%s], ctxHandle[%p]",
61 : __func__,
62 : devPhyId,
63 : ipAddr.Describe().c_str(),
64 : ctxHandle_);
65 :
66 23 : cacheKey_ = MemMgrCacheKey{devPhyId, COMM_PROTOCOL_UBC_CTP, ipAddr, LocTypeToPortType(endpointDesc_.loc.locType)};
67 23 : auto &cache = ProcRegedMemMgrCache::GetInstance();
68 46 : EXCEPTION_CATCH(regedMemMgr_ = cache.GetOrCreate(cacheKey_, [this]() {
69 : auto m = std::make_shared<UbRegedMemMgr>();
70 : m->rdmaHandle_ = this->ctxHandle_;
71 : return m;
72 : }), return HCCL_E_PARA);
73 23 : return HCCL_SUCCESS;
74 : }
75 :
76 1 : HcclResult CpuUrmaEndpoint::ServerSocketListen(const uint32_t port)
77 : {
78 1 : Hccl::IpAddress ipAddr{};
79 1 : CHK_RET(CommAddrToIpAddress(endpointDesc_.commAddr, ipAddr));
80 :
81 1 : s32 devId = 0;
82 1 : CHK_RET(hrtGetDevice(&devId));
83 1 : u32 devPhyId = 0;
84 1 : CHK_RET(hrtGetDevicePhyIdByIndex(devId, devPhyId));
85 :
86 1 : Hccl::DevNetPortType type = Hccl::DevNetPortType(Hccl::ConnectProtoType::UB);
87 1 : Hccl::PortData localPort = Hccl::PortData(devPhyId, type, 0, ipAddr);
88 :
89 1 : HCCL_INFO("[CpuUrmaEndpoint::%s] devicePhyId[%u] ipAddress[%s]",
90 : __func__, devPhyId, ipAddr.Describe().c_str());
91 :
92 1 : uint32_t requestPort = port;
93 1 : CHK_RET(ServerSocketManager::GetInstance().ServerSocketStartListen(localPort, Hccl::NicType::HOST_NIC_TYPE, devPhyId, &requestPort));
94 :
95 1 : return HCCL_SUCCESS;
96 : }
97 :
98 2 : inline HcclResult CpuUrmaEndpoint::ServerSocketStopListenImpl(const uint32_t port)
99 : {
100 2 : Hccl::IpAddress ipAddr{};
101 2 : CHK_RET(CommAddrToIpAddress(endpointDesc_.commAddr, ipAddr));
102 :
103 2 : s32 devId = 0;
104 2 : CHK_RET(hrtGetDevice(&devId));
105 2 : u32 devPhyId = 0;
106 2 : CHK_RET(hrtGetDevicePhyIdByIndex(devId, devPhyId));
107 2 : Hccl::DevNetPortType type = Hccl::DevNetPortType(Hccl::ConnectProtoType::UB);
108 2 : Hccl::PortData localPort = Hccl::PortData(devPhyId, type, 0, ipAddr);
109 2 : CHK_RET(ServerSocketManager::GetInstance().ServerSocketStopListen(localPort, Hccl::NicType::HOST_NIC_TYPE, port));
110 :
111 1 : return HCCL_SUCCESS;
112 : }
113 :
114 1 : HcclResult CpuUrmaEndpoint::ServerSocketStopListen(const uint32_t port)
115 : {
116 1 : return ServerSocketStopListenImpl(port);
117 : }
118 :
119 1 : HcclResult CpuUrmaEndpoint::ServerSocketGetListenPort(uint32_t *port)
120 : {
121 1 : std::lock_guard<std::mutex> lock(portMutex_);
122 1 : Hccl::IpAddress localIpAddr{};
123 1 : CHK_RET(CommAddrToIpAddress(endpointDesc_.commAddr, localIpAddr));
124 :
125 1 : s32 deviceId = 0;
126 1 : CHK_RET(hrtGetDevice(&deviceId));
127 1 : u32 devicePhyId = 0;
128 1 : CHK_RET(hrtGetDevicePhyIdByIndex(deviceId, devicePhyId));
129 :
130 1 : Hccl::DevNetPortType portType = Hccl::DevNetPortType(Hccl::ConnectProtoType::UB);
131 1 : Hccl::PortData portData = Hccl::PortData(devicePhyId, portType, 0, localIpAddr);
132 :
133 1 : HCCL_INFO("[CpuUrmaEndpoint::%s] devicePhyId[%u] ipAddress[%s]",
134 : __func__, devicePhyId, localIpAddr.Describe().c_str());
135 :
136 : // 已有监听端口则直接返回
137 1 : if (dynamicPort_ != HCCL_INVALID_PORT) {
138 0 : *port = dynamicPort_;
139 0 : HCCL_INFO("[CpuUrmaEndpoint::%s] already listening, return existing port[%u]", __func__, dynamicPort_);
140 0 : return HCCL_SUCCESS;
141 : }
142 :
143 1 : uint32_t requestPort = 0;
144 1 : CHK_RET(ServerSocketManager::GetInstance().ServerSocketStartListen(portData, Hccl::NicType::HOST_NIC_TYPE, devicePhyId, &requestPort));
145 1 : if (requestPort == 0 || requestPort == HCCL_INVALID_PORT) {
146 0 : HCCL_ERROR("[CpuUrmaEndpoint::%s] get listen port failed, port is invalid.", __func__);
147 0 : return HCCL_E_NETWORK;
148 : }
149 1 : dynamicPort_ = requestPort;
150 1 : *port = dynamicPort_;
151 :
152 1 : return HCCL_SUCCESS;
153 1 : }
154 :
155 1 : HcclResult CpuUrmaEndpoint::RegisterMemory(HcommMem mem, const char *memTag, void **memHandle)
156 : {
157 1 : CHK_RET(this->regedMemMgr_->RegisterMemory(mem, memTag, memHandle));
158 1 : return HCCL_SUCCESS;
159 : }
160 :
161 1 : HcclResult CpuUrmaEndpoint::UnregisterMemory(void* memHandle)
162 : {
163 1 : CHK_RET(this->regedMemMgr_->UnregisterMemory(memHandle));
164 1 : return HCCL_SUCCESS;
165 : }
166 :
167 0 : HcclResult CpuUrmaEndpoint::MemoryExport(void *memHandle, void **memDesc, uint32_t *memDescLen)
168 : {
169 0 : CHK_RET(this->regedMemMgr_->MemoryExport(this->endpointDesc_, memHandle, memDesc, memDescLen));
170 0 : return HCCL_SUCCESS;
171 : }
172 :
173 0 : HcclResult CpuUrmaEndpoint::MemoryImport(const void *memDesc, uint32_t descLen, HcommMem *outMem)
174 : {
175 0 : CHK_RET(this->regedMemMgr_->MemoryImport(memDesc, descLen, outMem));
176 0 : return HCCL_SUCCESS;
177 : }
178 :
179 0 : HcclResult CpuUrmaEndpoint::MemoryUnimport(const void *memDesc, uint32_t descLen)
180 : {
181 0 : CHK_RET(this->regedMemMgr_->MemoryUnimport(memDesc, descLen));
182 0 : return HCCL_SUCCESS;
183 : }
184 :
185 0 : HcclResult CpuUrmaEndpoint::GetAllMemHandles(void **memHandles, uint32_t *memHandleNum)
186 : {
187 0 : CHK_RET(this->regedMemMgr_->GetAllMemHandles(memHandles, memHandleNum));
188 0 : return HCCL_SUCCESS;
189 : }
190 :
191 : }
|
