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 :
19 : #include "hcom_common.h"
20 : #include "exception_handler.h"
21 :
22 : namespace hcomm {
23 :
24 11 : EndpointPair::~EndpointPair()
25 : {
26 20 : for (auto &channels : channelHandles_) {
27 9 : if (channels.second.empty()) {
28 3 : continue;
29 : }
30 6 : (void)ChannelProcess::ChannelDestroy(channels.second.data(), channels.second.size());
31 : }
32 11 : }
33 :
34 10 : HcclResult EndpointPair::Init()
35 : {
36 10 : EXCEPTION_CATCH(socketMgr_ = std::make_unique<SocketMgr>(), return HCCL_E_PTR);
37 10 : channelHandles_.clear();
38 : s32 devLogicId;
39 10 : CHK_RET(hrtGetDevice(&devLogicId));
40 10 : CHK_RET(hrtGetDevicePhyIdByIndex(static_cast<u32>(devLogicId), devicePhyId_));
41 :
42 10 : return HCCL_SUCCESS;
43 : }
44 :
45 3 : HcclResult EndpointPair::GetHostSocketWithRank(const uint32_t myRank, const uint32_t rmtRank, const std::string &socketTag,
46 : const uint32_t listenPort, u32 reuseIdx, Hccl::Socket*& socket)
47 : {
48 3 : uint32_t connectMode = 0;
49 3 : Hccl::LinkData linkData = BuildDefaultLinkData();
50 3 : CHK_RET(EndpointDescPairToLinkData(localEndpointDesc_, remoteEndpointDesc_, linkData, reuseIdx));
51 3 : std::string linkTag = socketTag;
52 3 : if (linkData.GetReuseIdx() != "0") {
53 0 : linkTag += ("_" + linkData.GetReuseIdx());
54 : }
55 :
56 : DevType devType;
57 3 : CHK_RET(hrtGetDeviceType(devType));
58 3 : if (devType == DevType::DEV_TYPE_910B && localEndpointDesc_.loc.locType != remoteEndpointDesc_.loc.locType) {
59 0 : connectMode = 1;
60 : }
61 :
62 : /* A2: host nic(cpu roce channel) -- device nic(transport ibv)时，两边ip地址格式不一样，判断大小算法不匹配
63 : * 修改成按照rank id大小判断server和client */
64 3 : Hccl::SocketConfig socketConfig = Hccl::SocketConfig(linkData, listenPort, linkTag, connectMode, myRank, rmtRank);
65 3 : CHK_RET(socketMgr_->GetHostSocket(socketConfig, socket));
66 3 : return HCCL_SUCCESS;
67 3 : }
68 :
69 6 : HcclResult EndpointPair::EnsureSocketMgrCompat(const uint32_t myRank, const std::string &socketTag)
70 : {
71 6 : if (!socketMgrCompat_) {
72 2 : int32_t devLogicId = HcclGetThreadDeviceId();
73 2 : uint32_t devPhyId{0};
74 2 : CHK_RET(hrtGetDevicePhyIdByIndex(static_cast<uint32_t>(devLogicId), devPhyId));
75 2 : EXCEPTION_CATCH(socketMgrCompat_ =
76 : std::make_unique<Hccl::SocketManager>(myRank, devPhyId, devLogicId, socketTag),
77 : return HCCL_E_PTR);
78 2 : CHK_PTR_NULL(rankIpPortMap_);
79 2 : socketMgrCompat_->SetDeviceServerListenPortMap(*rankIpPortMap_);
80 : }
81 6 : return HCCL_SUCCESS;
82 : }
83 :
84 12 : Hccl::SocketConfig EndpointPair::BuildSocketConfig(const Hccl::LinkData &linkData, const std::string &socketTag)
85 : {
86 12 : std::string linkTag = socketTag;
87 12 : if (linkData.GetReuseIdx() != "0") {
88 2 : linkTag += ("_" + linkData.GetReuseIdx());
89 : }
90 24 : return Hccl::SocketConfig(linkData.GetRemoteRankId(), linkData, linkTag);
91 12 : }
92 :
93 9 : HcclResult EndpointPair::HandleHostSocketOrBuildLinkData(const uint32_t myRank, const uint32_t rmtRank,
94 : const std::string &socketTag, u32 reuseIdx, const uint32_t listenPort, Hccl::Socket*& socket,
95 : uint32_t devicePhyId, uint32_t remoteDevicePhyId, Hccl::LinkData &linkData, bool &isHost)
96 : {
97 9 : if (localEndpointDesc_.loc.locType == EndpointLocType::ENDPOINT_LOC_TYPE_HOST) {
98 3 : std::string socketTagPrefix = socketTag;
99 3 : if (myRank <= rmtRank) {
100 2 : socketTagPrefix += "_" + std::to_string(myRank) + "_" + std::to_string(rmtRank);
101 : } else {
102 1 : socketTagPrefix += "_" + std::to_string(rmtRank) + "_" + std::to_string(myRank);
103 : }
104 3 : CHK_RET(this->GetHostSocketWithRank(myRank, rmtRank, socketTagPrefix, listenPort, reuseIdx, socket));
105 3 : isHost = true;
106 3 : return HCCL_SUCCESS;
107 3 : }
108 6 : isHost = false;
109 6 : CHK_RET(EndpointDescPairToLinkDataWithRankIds(myRank, rmtRank,
110 : localEndpointDesc_, remoteEndpointDesc_, linkData, devicePhyId, remoteDevicePhyId, reuseIdx));
111 6 : return HCCL_SUCCESS;
112 : }
113 :
114 9 : HcclResult EndpointPair::GetSocketInternal(const uint32_t myRank, const uint32_t rmtRank,
115 : const std::string &socketTag, u32 reuseIdx, const uint32_t listenPort, Hccl::Socket*& socket,
116 : uint32_t devicePhyId, uint32_t remoteDevicePhyId, bool connectMode)
117 : {
118 9 : Hccl::LinkData linkData = BuildDefaultLinkData();
119 9 : bool isHost = false;
120 9 : CHK_RET(HandleHostSocketOrBuildLinkData(myRank, rmtRank, socketTag, reuseIdx, listenPort, socket,
121 : devicePhyId, remoteDevicePhyId, linkData, isHost));
122 9 : if (isHost) {
123 3 : return HCCL_SUCCESS;
124 : }
125 : EXCEPTION_HANDLE_BEGIN
126 6 : Hccl::SocketConfig socketConfig = BuildSocketConfig(linkData, socketTag);
127 6 : if (connectMode) {
128 6 : CHK_PTR_NULL(socketMgrCompat_);
129 6 : socketMgrCompat_->ConnectSockets(socketConfig);
130 : } else {
131 0 : CHK_RET(EnsureSocketMgrCompat(myRank, socketTag));
132 0 : socketMgrCompat_->BatchCreateSockets(socketConfig);
133 : }
134 6 : socket = socketMgrCompat_->GetConnectedSocket(socketConfig);
135 6 : CHK_PTR_NULL(socket);
136 6 : EXCEPTION_HANDLE_END
137 6 : return HCCL_SUCCESS;
138 : }
139 :
140 7 : HcclResult EndpointPair::ServerInit(const uint32_t myRank, const uint32_t rmtRank,
141 : const std::string &socketTag, u32 reuseIdx, uint32_t devicePhyId, uint32_t remoteDevicePhyId)
142 : {
143 7 : if (localEndpointDesc_.loc.locType == EndpointLocType::ENDPOINT_LOC_TYPE_HOST) {
144 : // host网卡不走device的socket监听
145 1 : return HCCL_SUCCESS;
146 : }
147 : // server监听
148 6 : Hccl::LinkData linkData = BuildDefaultLinkData();
149 6 : CHK_RET(EndpointDescPairToLinkDataWithRankIds(myRank, rmtRank, localEndpointDesc_,
150 : remoteEndpointDesc_, linkData, devicePhyId, remoteDevicePhyId, reuseIdx));
151 : EXCEPTION_HANDLE_BEGIN
152 6 : CHK_RET(EnsureSocketMgrCompat(myRank, socketTag));
153 6 : Hccl::SocketConfig socketConfig = BuildSocketConfig(linkData, socketTag);
154 : // 调用sock的server监听接口
155 6 : socketMgrCompat_->ServerListen(socketConfig);
156 6 : EXCEPTION_HANDLE_END
157 :
158 6 : return HCCL_SUCCESS;
159 : }
160 :
161 7 : HcclResult EndpointPair::GetConnectedSocket(const uint32_t myRank, const uint32_t rmtRank,
162 : const std::string &socketTag, u32 reuseIdx, const uint32_t listenPort, Hccl::Socket*& socket, uint32_t devicePhyId, uint32_t remoteDevicePhyId)
163 : {
164 : // 该接口内进行建链和获取socket
165 7 : return GetSocketInternal(myRank, rmtRank, socketTag, reuseIdx, listenPort, socket,
166 7 : devicePhyId, remoteDevicePhyId, true);
167 : }
168 :
169 2 : HcclResult EndpointPair::GetSocket(const uint32_t myRank, const uint32_t rmtRank,
170 : const std::string &socketTag, u32 reuseIdx, const uint32_t listenPort, Hccl::Socket*& socket, uint32_t devicePhyId, uint32_t remoteDevicePhyId)
171 : {
172 : // 临时方案：支持混跑新增，非Roce场景走orion socketMgr实现server socket复用
173 2 : return GetSocketInternal(myRank, rmtRank, socketTag, reuseIdx, listenPort, socket,
174 2 : devicePhyId, remoteDevicePhyId, false);
175 : }
176 :
177 18 : HcclResult EndpointPair::CreateChannel(EndpointHandle endpointHandle, CommEngine engine, u32 reuseIdx,
178 : HcommChannelDesc *channelDescs, ChannelHandle *channels)
179 : {
180 18 : if (channelHandles_.find(engine) == channelHandles_.end() || channelHandles_[engine].size() <= reuseIdx) {
181 12 : CHK_RET_UNAVAIL(static_cast<HcclResult>(
182 : HcommCollectiveChannelCreate(endpointHandle, engine, channelDescs, 1, channels)));
183 10 : channelHandles_[engine].push_back(channels[0]);
184 10 : return HCCL_SUCCESS;
185 : }
186 :
187 6 : channels[0] = channelHandles_[engine][reuseIdx];
188 6 : if (channelDescs->memHandleNum > 1) {
189 0 : CHK_RET(static_cast<HcclResult>(HcommChannelUpdateMemInfo(channelDescs->memHandles + 1, channelDescs->memHandleNum - 1, channels[0])));
190 : }
191 6 : return HCCL_SUCCESS;
192 : }
193 :
194 : // 找到对应的channelhandle，调用HcommChannelDestroy销毁平台层对象，并删除channelHandles_中的channelHandle元素
195 6 : HcclResult EndpointPair::DestroyChannel(CommEngine engine, u32 reuseIdx)
196 : {
197 6 : if (IsChannelNotExist(engine, reuseIdx)) {
198 1 : HCCL_WARNING("EndpointPair::DestroyChannel: engine[%s] reuseIdx[%u], channelHandle size[%u],"
199 : "channel not found, skip destroy channel", GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), reuseIdx, channelHandles_[engine].size());
200 1 : return HCCL_SUCCESS;
201 : }
202 5 : HCCL_INFO("EndpointPair::DestroyChannel: engine[%s] reuseIdx[%u], channelHandle size[%u],"
203 : "start destroy channel", GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), reuseIdx, channelHandles_[engine].size());
204 5 : ChannelHandle channelHandle = channelHandles_[engine][reuseIdx];
205 5 : CHK_RET(static_cast<HcclResult>(HcommChannelDestroy(&channelHandle, 1)));
206 : // 去掉channelHandles_中reuseIdx位置的channelHandle
207 5 : channelHandles_[engine].erase(channelHandles_[engine].begin() + reuseIdx);
208 5 : HCCL_INFO("EndpointPair::DestroyChannel: engine[%s] reuseIdx[%u] destroy channel success,"
209 : "channelHandle size[%u]", GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), reuseIdx, channelHandles_[engine].size());
210 5 : return HCCL_SUCCESS;
211 : }
212 :
213 : // 检查channel是否存在，channel不存在则返回true
214 24 : bool EndpointPair::IsChannelNotExist(CommEngine engine, u32 reuseIdx)
215 : {
216 24 : return channelHandles_.find(engine) == channelHandles_.end() || channelHandles_[engine].size() <= reuseIdx;
217 : }
218 :
219 1 : const std::unordered_map<CommEngine, std::vector<ChannelHandle>>& EndpointPair::GetChannelHandles()
220 : {
221 1 : return channelHandles_;
222 : }
223 :
224 : } // namespace hcomm
|
