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 "endpoint_pair.h"
12 : #include "socket_config.h"
13 : #include "hcomm_c_adpt.h"
14 : #include "orion_adpt_utils.h"
15 : #include "channel_process.h"
16 : #include "comm_engine_utils.h"
17 :
18 : #include "hcom_common.h"
19 : #include "exception_handler.h"
20 :
21 : namespace hcomm {
22 :
23 39 : EndpointPair::~EndpointPair()
24 : {
25 78 : for (auto& channels : channelHandles_) {
26 39 : if (channels.second.empty()) {
27 17 : continue;
28 : }
29 22 : (void)ChannelProcess::ChannelDestroy(channels.second.data(), channels.second.size());
30 : }
31 39 : }
32 :
33 38 : HcclResult EndpointPair::Init()
34 : {
35 38 : std::lock_guard<std::mutex> lock(channelMtx_);
36 38 : EXCEPTION_CATCH(socketMgr_ = std::make_unique<SocketMgr>(), return HCCL_E_PTR);
37 38 : channelHandles_.clear();
38 : s32 devLogicId;
39 38 : CHK_RET(hrtGetDevice(&devLogicId));
40 38 : CHK_RET(hrtGetDevicePhyIdByIndex(static_cast<u32>(devLogicId), devicePhyId_));
41 :
42 38 : return HCCL_SUCCESS;
43 38 : }
44 :
45 3 : HcclResult EndpointPair::GetHostSocketWithRank(
46 : const uint32_t myRank, const uint32_t rmtRank, const std::string& socketTag, const uint32_t listenPort,
47 : u32 reuseIdx, Hccl::Socket*& socket)
48 : {
49 3 : uint32_t connectMode = 0;
50 3 : Hccl::LinkData linkData = BuildDefaultLinkData();
51 3 : CHK_RET(EndpointDescPairToLinkData(localEndpointDesc_, remoteEndpointDesc_, linkData, reuseIdx));
52 3 : std::string linkTag = socketTag;
53 3 : if (linkData.GetReuseIdx() != "0") {
54 0 : linkTag += ("_" + linkData.GetReuseIdx());
55 : }
56 :
57 : DevType devType;
58 3 : CHK_RET(hrtGetDeviceType(devType));
59 3 : if (devType == DevType::DEV_TYPE_910B && localEndpointDesc_.loc.locType != remoteEndpointDesc_.loc.locType) {
60 0 : connectMode = 1;
61 : }
62 :
63 : /* A2: host nic(cpu roce channel) -- device nic(transport ibv)时，两边ip地址格式不一样，判断大小算法不匹配
64 : * 修改成按照rank id大小判断server和client */
65 3 : Hccl::SocketConfig socketConfig = Hccl::SocketConfig(linkData, listenPort, linkTag, connectMode, myRank, rmtRank);
66 3 : CHK_RET(socketMgr_->GetHostSocket(socketConfig, socket));
67 3 : return HCCL_SUCCESS;
68 3 : }
69 :
70 20 : HcclResult EndpointPair::EnsureSocketMgrCompat(const uint32_t myRank, const std::string& socketTag)
71 : {
72 : {
73 20 : std::lock_guard<std::mutex> lock(socketMgrMtx_);
74 20 : if (socketMgrCompat_) {
75 12 : return HCCL_SUCCESS;
76 : }
77 20 : }
78 :
79 8 : int32_t devLogicId = HcclGetThreadDeviceId();
80 8 : uint32_t devPhyId{0};
81 8 : CHK_RET(hrtGetDevicePhyIdByIndex(static_cast<uint32_t>(devLogicId), devPhyId));
82 8 : std::unique_ptr<Hccl::SocketManager> newMgr = nullptr;
83 8 : EXCEPTION_CATCH(
84 : newMgr = std::make_unique<Hccl::SocketManager>(myRank, devPhyId, devLogicId, socketTag), return HCCL_E_PTR);
85 8 : CHK_PTR_NULL(rankIpPortMap_);
86 8 : newMgr->SetDeviceServerListenPortMap(*rankIpPortMap_);
87 :
88 : {
89 8 : std::lock_guard<std::mutex> lock(socketMgrMtx_);
90 8 : if (socketMgrCompat_) {
91 0 : return HCCL_SUCCESS;
92 : }
93 8 : socketMgrCompat_ = std::move(newMgr);
94 8 : }
95 :
96 8 : return HCCL_SUCCESS;
97 8 : }
98 :
99 40 : Hccl::SocketConfig EndpointPair::BuildSocketConfig(const Hccl::LinkData& linkData, const std::string& socketTag)
100 : {
101 40 : std::string linkTag = socketTag;
102 40 : if (linkData.GetReuseIdx() != "0") {
103 6 : linkTag += ("_" + linkData.GetReuseIdx());
104 : }
105 80 : return Hccl::SocketConfig(linkData.GetRemoteRankId(), linkData, linkTag);
106 40 : }
107 :
108 23 : HcclResult EndpointPair::HandleHostSocketOrBuildLinkData(
109 : const uint32_t myRank, const uint32_t rmtRank, const std::string& socketTag, u32 reuseIdx,
110 : const uint32_t listenPort, Hccl::Socket*& socket, uint32_t devicePhyId, uint32_t remoteDevicePhyId,
111 : Hccl::LinkData& linkData, bool& isHost)
112 : {
113 23 : if (localEndpointDesc_.loc.locType == EndpointLocType::ENDPOINT_LOC_TYPE_HOST) {
114 3 : std::string socketTagPrefix = socketTag;
115 3 : if (myRank <= rmtRank) {
116 2 : socketTagPrefix += "_" + std::to_string(myRank) + "_" + std::to_string(rmtRank);
117 : } else {
118 1 : socketTagPrefix += "_" + std::to_string(rmtRank) + "_" + std::to_string(myRank);
119 : }
120 3 : CHK_RET(this->GetHostSocketWithRank(myRank, rmtRank, socketTagPrefix, listenPort, reuseIdx, socket));
121 3 : isHost = true;
122 3 : return HCCL_SUCCESS;
123 3 : }
124 20 : isHost = false;
125 20 : CHK_RET(EndpointDescPairToLinkDataWithRankIds(
126 : myRank, rmtRank, localEndpointDesc_, remoteEndpointDesc_, linkData, devicePhyId, remoteDevicePhyId, reuseIdx));
127 20 : return HCCL_SUCCESS;
128 : }
129 :
130 23 : HcclResult EndpointPair::GetSocketInternal(
131 : const uint32_t myRank, const uint32_t rmtRank, const std::string& socketTag, u32 reuseIdx,
132 : const uint32_t listenPort, Hccl::Socket*& socket, uint32_t devicePhyId, uint32_t remoteDevicePhyId,
133 : bool connectMode)
134 : {
135 23 : Hccl::LinkData linkData = BuildDefaultLinkData();
136 23 : bool isHost = false;
137 23 : CHK_RET(HandleHostSocketOrBuildLinkData(
138 : myRank, rmtRank, socketTag, reuseIdx, listenPort, socket, devicePhyId, remoteDevicePhyId, linkData, isHost));
139 23 : if (isHost) {
140 3 : return HCCL_SUCCESS;
141 : }
142 : EXCEPTION_HANDLE_BEGIN
143 20 : Hccl::SocketConfig socketConfig = BuildSocketConfig(linkData, socketTag);
144 20 : if (connectMode) {
145 20 : CHK_PTR_NULL(socketMgrCompat_);
146 20 : socketMgrCompat_->ConnectSockets(socketConfig);
147 : } else {
148 0 : CHK_RET(EnsureSocketMgrCompat(myRank, socketTag));
149 0 : socketMgrCompat_->BatchCreateSockets(socketConfig);
150 : }
151 20 : socket = socketMgrCompat_->GetConnectedSocket(socketConfig);
152 20 : CHK_PTR_NULL(socket);
153 20 : EXCEPTION_HANDLE_END
154 20 : return HCCL_SUCCESS;
155 : }
156 :
157 21 : HcclResult EndpointPair::ServerInit(
158 : const uint32_t myRank, const uint32_t rmtRank, const std::string& socketTag, u32 reuseIdx, uint32_t devicePhyId,
159 : uint32_t remoteDevicePhyId)
160 : {
161 21 : if (localEndpointDesc_.loc.locType == EndpointLocType::ENDPOINT_LOC_TYPE_HOST) {
162 : // host网卡不走device的socket监听
163 1 : return HCCL_SUCCESS;
164 : }
165 : // server监听
166 20 : Hccl::LinkData linkData = BuildDefaultLinkData();
167 20 : CHK_RET(EndpointDescPairToLinkDataWithRankIds(
168 : myRank, rmtRank, localEndpointDesc_, remoteEndpointDesc_, linkData, devicePhyId, remoteDevicePhyId, reuseIdx));
169 : EXCEPTION_HANDLE_BEGIN
170 20 : CHK_RET(EnsureSocketMgrCompat(myRank, socketTag));
171 20 : Hccl::SocketConfig socketConfig = BuildSocketConfig(linkData, socketTag);
172 : // 调用sock的server监听接口
173 20 : socketMgrCompat_->ServerListen(socketConfig);
174 20 : EXCEPTION_HANDLE_END
175 :
176 20 : return HCCL_SUCCESS;
177 : }
178 :
179 21 : HcclResult EndpointPair::GetConnectedSocket(
180 : const uint32_t myRank, const uint32_t rmtRank, const std::string& socketTag, u32 reuseIdx,
181 : const uint32_t listenPort, Hccl::Socket*& socket, uint32_t devicePhyId, uint32_t remoteDevicePhyId)
182 : {
183 : // 该接口内进行建链和获取socket
184 21 : return GetSocketInternal(
185 21 : myRank, rmtRank, socketTag, reuseIdx, listenPort, socket, devicePhyId, remoteDevicePhyId, true);
186 : }
187 :
188 2 : HcclResult EndpointPair::GetSocket(
189 : const uint32_t myRank, const uint32_t rmtRank, const std::string& socketTag, u32 reuseIdx,
190 : const uint32_t listenPort, Hccl::Socket*& socket, uint32_t devicePhyId, uint32_t remoteDevicePhyId)
191 : {
192 : // 临时方案：支持混跑新增，非Roce场景走orion socketMgr实现server socket复用
193 2 : return GetSocketInternal(
194 2 : myRank, rmtRank, socketTag, reuseIdx, listenPort, socket, devicePhyId, remoteDevicePhyId, false);
195 : }
196 :
197 34 : HcclResult EndpointPair::CreateChannel(
198 : EndpointHandle endpointHandle, CommEngine engine, u32 reuseIdx, HcommChannelDesc* channelDescs,
199 : ChannelHandle* channels)
200 : {
201 34 : std::lock_guard<std::mutex> lock(channelMtx_);
202 34 : if (channelHandles_.find(engine) == channelHandles_.end() || channelHandles_[engine].size() <= reuseIdx) {
203 24 : CHK_RET_UNAVAIL(
204 : static_cast<HcclResult>(HcommCollectiveChannelCreate(endpointHandle, engine, channelDescs, 1, channels)));
205 22 : channelHandles_[engine].push_back(channels[0]);
206 : // 记录真实槽位下标：UNREUSE 通道的入参 reuseIdx 为 0xFFFFFFFF，实际槽位是 push_back 后的下标
207 22 : handleToLoc_[channels[0]] = {engine, static_cast<u32>(channelHandles_[engine].size() - 1)};
208 22 : return HCCL_SUCCESS;
209 : }
210 :
211 10 : channels[0] = channelHandles_[engine][reuseIdx];
212 10 : if (channelDescs->memHandleNum > 1) {
213 0 : CHK_RET(static_cast<HcclResult>(
214 : HcommChannelUpdateMemInfo(channelDescs->memHandles + 1, channelDescs->memHandleNum - 1, channels[0])));
215 : }
216 10 : return HCCL_SUCCESS;
217 34 : }
218 :
219 : // 找到对应的channelhandle，调用HcommChannelDestroy销毁平台层对象，并删除channelHandles_中的channelHandle元素
220 30 : HcclResult EndpointPair::DestroyChannel(CommEngine engine, u32 reuseIdx)
221 : {
222 30 : std::lock_guard<std::mutex> lock(channelMtx_);
223 30 : if (channelHandles_.find(engine) == channelHandles_.end() || channelHandles_[engine].size() <= reuseIdx) {
224 1 : HCCL_WARNING(
225 : "EndpointPair::DestroyChannel: engine[%s] reuseIdx[%u], channelHandle size[%u],"
226 : "channel not found, skip destroy channel",
227 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), reuseIdx, channelHandles_[engine].size());
228 1 : return HCCL_SUCCESS;
229 : }
230 29 : HCCL_INFO(
231 : "EndpointPair::DestroyChannel: engine[%s] reuseIdx[%u], channelHandle size[%u],"
232 : "start destroy channel",
233 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), reuseIdx, channelHandles_[engine].size());
234 29 : ChannelHandle channelHandle = channelHandles_[engine][reuseIdx];
235 : // 无论 HcommChannelDestroy 成功与否，底层 channel 对象已被从全局 map 移除（channel 不可用），
236 : // host 侧索引必须同步清理，避免后续复用到失效 handle
237 29 : HcclResult destroyRet = static_cast<HcclResult>(HcommChannelDestroy(&channelHandle, 1));
238 29 : if (destroyRet != HCCL_SUCCESS) {
239 2 : HCCL_WARNING(
240 : "EndpointPair::DestroyChannel: HcommChannelDestroy failed, ret[%d], still clean host index.", destroyRet);
241 : }
242 : // 先删反查索引再 erase 向量: erase 会使后续元素下标前移
243 29 : handleToLoc_.erase(channelHandle);
244 : // 去掉channelHandles_中reuseIdx位置的channelHandle
245 29 : channelHandles_[engine].erase(channelHandles_[engine].begin() + reuseIdx);
246 : // 同 engine 后续 handle 因 erase 下标前移, 需同步修正反查索引
247 29 : auto& handlesVec = channelHandles_[engine];
248 41 : for (u32 idx = reuseIdx; idx < handlesVec.size(); ++idx) {
249 12 : auto locIt = handleToLoc_.find(handlesVec[idx]);
250 12 : if (locIt != handleToLoc_.end()) {
251 10 : locIt->second.second = idx;
252 : }
253 : }
254 29 : HCCL_INFO(
255 : "EndpointPair::DestroyChannel: engine[%s] reuseIdx[%u] destroy channel success,"
256 : "channelHandle size[%u]",
257 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), reuseIdx, channelHandles_[engine].size());
258 29 : return destroyRet;
259 30 : }
260 :
261 : // 检查channel是否存在，channel不存在则返回true
262 34 : bool EndpointPair::IsChannelNotExist(CommEngine engine, u32 reuseIdx)
263 : {
264 34 : std::lock_guard<std::mutex> lock(channelMtx_);
265 68 : return channelHandles_.find(engine) == channelHandles_.end() || channelHandles_[engine].size() <= reuseIdx;
266 34 : }
267 :
268 1 : std::unordered_map<CommEngine, std::vector<ChannelHandle>> EndpointPair::GetChannelHandles() const
269 : {
270 1 : std::lock_guard<std::mutex> lock(channelMtx_);
271 2 : return channelHandles_;
272 1 : }
273 :
274 18 : bool EndpointPair::GetChannelHandle(CommEngine engine, u32 reuseIdx, ChannelHandle& handle) const
275 : {
276 18 : std::lock_guard<std::mutex> lock(channelMtx_);
277 18 : auto it = channelHandles_.find(engine);
278 18 : if (it == channelHandles_.end() || reuseIdx >= it->second.size()) {
279 6 : return false;
280 : }
281 12 : handle = it->second[reuseIdx];
282 12 : return true;
283 18 : }
284 :
285 30 : bool EndpointPair::FindChannelLoc(ChannelHandle handle, CommEngine& engine, u32& reuseIdx) const
286 : {
287 30 : std::lock_guard<std::mutex> lock(channelMtx_);
288 30 : auto it = handleToLoc_.find(handle);
289 30 : if (it == handleToLoc_.end()) {
290 1 : return false;
291 : }
292 29 : engine = it->second.first;
293 29 : reuseIdx = it->second.second;
294 29 : return true;
295 30 : }
296 :
297 : } // namespace hcomm
|
