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