Line data Source code
1 : /**
2 : * Copyright (c) 2026 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 "exchange_info_mgr.h"
11 : #include "env_config/env_config_v2.h"
12 : #include "rank_consistency_checker_v2.h"
13 : #include "hcomm_res_defs.h"
14 : #include "hcomm_channel.h"
15 : #include "comm_engine_utils.h"
16 :
17 : namespace hccl {
18 230 : ExchangeInfoMgr::ExchangeInfoMgr() {}
19 :
20 230 : ExchangeInfoMgr::~ExchangeInfoMgr()
21 : {
22 230 : HCCL_INFO("[ExchangeInfoMgr][~ExchangeInfoMgr] CollCommConfigConsistency deinit");
23 230 : }
24 :
25 2 : HcclResult ExchangeInfoMgr::BatchExchangeAndCheckConsistency(
26 : const HcclChannelDesc* channelDescs, const std::vector<HcommChannelDesc>& hcommDescs, uint32_t channelNum,
27 : const std::vector<std::pair<u32, u32>>& newChannels, CollCommConfigConsistency& collCommConfigConsistency,
28 : CommEngine engine) const
29 : {
30 2 : std::vector<Hccl::Socket*> sockets;
31 2 : std::vector<u32> remoteRanks;
32 2 : std::vector<HcommSocketRole> roles;
33 :
34 2 : if (channelNum == 0) {
35 0 : HCCL_INFO("[BatchExchangeAndCheckConsistency] channelNum is 0.");
36 0 : return HCCL_SUCCESS;
37 : }
38 :
39 4 : for (uint32_t i = 0; i < channelNum; i++) {
40 2 : u32 remoteRank = channelDescs[i].remoteRank;
41 2 : HcommSocket rawSocket = hcommDescs[i].socket;
42 2 : Hccl::Socket* socket = static_cast<Hccl::Socket*>(rawSocket);
43 2 : CHK_PRT_RET(
44 : socket == nullptr,
45 : HCCL_ERROR(
46 : "[BatchExchangeAndCheckConsistency] socket is null for channel[%u] remoteRank[%u].", i, remoteRank),
47 : HCCL_E_INTERNAL);
48 2 : sockets.push_back(socket);
49 2 : remoteRanks.push_back(remoteRank);
50 2 : roles.push_back(hcommDescs[i].role);
51 : }
52 :
53 2 : bool isHostLoc = (channelDescs[0].localEndpoint.loc.locType == ENDPOINT_LOC_TYPE_HOST);
54 2 : HCCL_INFO(
55 : "[BatchExchangeAndCheckConsistency] isHostLoc[%d], engine[%s]", isHostLoc,
56 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str());
57 :
58 : // host 网卡使用 Socket 同步收发接口
59 2 : if (isHostLoc) {
60 : // 校验HComm信息
61 1 : CHK_RET(CheckHcommInfo(channelDescs, hcommDescs, sockets, remoteRanks, roles, newChannels, false));
62 : // 交换HCCL算子信息
63 1 : CHK_RET(ExchangeUserInfo(sockets, remoteRanks, roles, collCommConfigConsistency));
64 : } else {
65 : // 校验HComm信息
66 1 : CHK_RET(CheckHcommInfo(channelDescs, hcommDescs, sockets, remoteRanks, roles, newChannels, true));
67 : // 交换HCCL算子信息
68 1 : CHK_RET(ExchangeUserInfoAsync(sockets, remoteRanks, roles, collCommConfigConsistency));
69 : }
70 2 : CHK_RET(collCommConfigConsistency.ResetExchangeInfo());
71 :
72 2 : return HCCL_SUCCESS;
73 2 : }
74 :
75 2 : HcclResult ExchangeInfoMgr::CheckHcommInfo(
76 : const HcclChannelDesc* channelDescs, const std::vector<HcommChannelDesc>& hcommDescs,
77 : const std::vector<Hccl::Socket*>& sockets, const std::vector<u32>& remoteRanks,
78 : const std::vector<HcommSocketRole>& roles, const std::vector<std::pair<u32, u32>>& newChannels, bool isAsync) const
79 : {
80 2 : s32 deviceLogicId = 0;
81 2 : (void)hrtGetDeviceRefresh(&deviceLogicId);
82 2 : auto& checker = RankConsistencyCheckerV2::GetInstance(deviceLogicId);
83 2 : u64 frameLenV2 = checker.GetCheckFrameLengthV2();
84 2 : std::vector<Hccl::Socket*> newSockets; // 新建链的
85 2 : std::vector<u32> newRemoteRanks;
86 2 : std::vector<HcommSocketRole> newRoles;
87 :
88 4 : for (const auto& newCh : newChannels) {
89 2 : u32 idx = newCh.first;
90 2 : u32 remoteRank = channelDescs[idx].remoteRank;
91 2 : HcommSocket rawSocket = hcommDescs[idx].socket;
92 2 : Hccl::Socket* socket = static_cast<Hccl::Socket*>(rawSocket);
93 2 : CHK_PRT_RET(
94 : socket == nullptr,
95 : HCCL_ERROR("[CheckHcommInfo] socket is null for channel[%u] remoteRank[%u].", idx, remoteRank),
96 : HCCL_E_INTERNAL);
97 2 : newSockets.push_back(socket);
98 2 : newRemoteRanks.push_back(remoteRank);
99 2 : newRoles.push_back(hcommDescs[idx].role);
100 : }
101 2 : if (newChannels.size() > 0) {
102 2 : checker.SetInconsistentCheckFirstDone(false);
103 : }
104 : // 只有rankConsistentState是first或者on时才进行hcomm信息校验
105 2 : int32_t mode = Hccl::EnvConfig::GetInstance().GetLogConfig().GetDfsConfig().rankConsistentState;
106 2 : bool isFirst = (mode == 0 && !checker.GetInconsistentCheckFirstDone()); // 0:first 1:on
107 2 : if (mode == 1 || isFirst) {
108 : // ====== 生成本端CheckFrameV2 ======
109 2 : CheckFrameV2 localFrame;
110 2 : CHK_RET(checker.GenerateCheckFrameV2(localFrame));
111 :
112 : // ====== 交换CheckFrameV2（定长，批量并发交换）======
113 2 : std::vector<CheckFrameV2> remoteFrames;
114 2 : u32 checkSocketSize = newSockets.size();
115 2 : if (isFirst) {
116 2 : remoteFrames.resize(newSockets.size());
117 2 : CHK_RET(BatchExchangeFixedData(
118 : newSockets, newRemoteRanks, newRoles, reinterpret_cast<const u8*>(&localFrame),
119 : static_cast<u32>(frameLenV2), reinterpret_cast<u8*>(remoteFrames.data()), static_cast<u32>(frameLenV2),
120 : isAsync));
121 : } else {
122 0 : checkSocketSize = sockets.size();
123 0 : remoteFrames.resize(sockets.size());
124 0 : CHK_RET(BatchExchangeFixedData(
125 : sockets, remoteRanks, roles, reinterpret_cast<const u8*>(&localFrame), static_cast<u32>(frameLenV2),
126 : reinterpret_cast<u8*>(remoteFrames.data()), static_cast<u32>(frameLenV2), isAsync));
127 : }
128 :
129 : // ====== 逐个比对CheckFrameV2（精确报错：环境变量名/子通信域参数名等）======
130 4 : for (u32 i = 0; i < checkSocketSize; i++) {
131 2 : CHK_RET(checker.CompareCheckFrameV2(localFrame, remoteFrames[i]));
132 : }
133 2 : HCCL_INFO("[BatchExchangeAndCheckConsistency] hcomm compare check suc.");
134 2 : checker.SetInconsistentCheckFirstDone(true);
135 2 : }
136 2 : return HCCL_SUCCESS;
137 2 : }
138 :
139 1 : HcclResult ExchangeInfoMgr::ExchangeUserInfo(
140 : const std::vector<Hccl::Socket*>& sockets, const std::vector<u32>& remoteRanks,
141 : const std::vector<HcommSocketRole>& roles, hccl::CollCommConfigConsistency& collCommConfigConsistency) const
142 : {
143 : (void)roles;
144 :
145 1 : u32 localExchangeInfoLen = collCommConfigConsistency.GetExchangeInfoLen();
146 1 : if (localExchangeInfoLen == 0) {
147 0 : HCCL_INFO("[ExchangeUserInfo] localExchangeInfoLen is 0.");
148 0 : return HCCL_SUCCESS;
149 : }
150 :
151 : // 交换infoLen
152 1 : std::vector<u32> remoteExchangeInfoLens(sockets.size(), 0);
153 1 : u8* recvData = reinterpret_cast<u8*>(remoteExchangeInfoLens.data());
154 1 : const u8* sendData = reinterpret_cast<const u8*>(&localExchangeInfoLen);
155 1 : u32 len = sizeof(u32);
156 2 : for (u32 i = 0; i < sockets.size(); i++) {
157 1 : sockets[i]->Send(sendData, len);
158 1 : sockets[i]->Recv(recvData + i * len, len);
159 : }
160 :
161 : // 交换info数据（长度可能不同，需逐个收发）
162 1 : std::vector<std::vector<u8>> remoteUserDatas(sockets.size());
163 2 : for (u32 i = 0; i < sockets.size(); i++) {
164 1 : std::vector<u8> exchangeBuf;
165 1 : collCommConfigConsistency.GetExchangeInfoBuf(exchangeBuf);
166 1 : remoteUserDatas[i].resize(remoteExchangeInfoLens[i], 0);
167 1 : sockets[i]->Send(exchangeBuf.data(), localExchangeInfoLen);
168 1 : sockets[i]->Recv(remoteUserDatas[i].data(), remoteExchangeInfoLens[i]);
169 1 : }
170 :
171 : // 存储对端交换信息
172 2 : for (u32 i = 0; i < sockets.size(); i++) {
173 1 : if (remoteExchangeInfoLens[i] > 0 && !remoteUserDatas[i].empty()) {
174 0 : CHK_RET(collCommConfigConsistency.StoreRemoteExchangeInfo(remoteRanks[i], remoteUserDatas[i]));
175 : }
176 : }
177 :
178 1 : HCCL_INFO("[ExchangeUserInfo] suc.");
179 1 : return HCCL_SUCCESS;
180 1 : }
181 :
182 1 : HcclResult ExchangeInfoMgr::ExchangeUserInfoAsync(
183 : const std::vector<Hccl::Socket*>& sockets, const std::vector<u32>& remoteRanks,
184 : const std::vector<HcommSocketRole>& roles, CollCommConfigConsistency& collCommConfigConsistency) const
185 : {
186 1 : u32 localExchangeInfoLen = collCommConfigConsistency.GetExchangeInfoLen();
187 1 : if (localExchangeInfoLen == 0) {
188 0 : HCCL_INFO("[ExchangeUserInfoAsync] localExchangeInfoLen is 0.");
189 0 : return HCCL_SUCCESS;
190 : }
191 : // 交换infoLen
192 1 : std::vector<u32> remoteExchangeInfoLens(sockets.size(), 0);
193 1 : CHK_RET(BatchExchangeFixedData(
194 : sockets, remoteRanks, roles, reinterpret_cast<const u8*>(&localExchangeInfoLen), sizeof(u32),
195 : reinterpret_cast<u8*>(remoteExchangeInfoLens.data()), sizeof(u32), true));
196 :
197 : // 交换info数据（两阶段异步收发，防死锁）
198 1 : std::vector<std::vector<u8>> remoteUserDatas(sockets.size());
199 : // 第一阶段：SERVER先Recv/CLIENT先Send
200 1 : CHK_RET(ExchangeAsyncDataPhase(
201 : sockets, roles, remoteRanks, remoteUserDatas, remoteExchangeInfoLens, localExchangeInfoLen,
202 : collCommConfigConsistency, true));
203 1 : CHK_RET(WaitActiveAsyncComplete(sockets, remoteRanks, roles, remoteExchangeInfoLens, localExchangeInfoLen, true));
204 : // 第二阶段：SERVER再Send/CLIENT再Recv
205 1 : CHK_RET(ExchangeAsyncDataPhase(
206 : sockets, roles, remoteRanks, remoteUserDatas, remoteExchangeInfoLens, localExchangeInfoLen,
207 : collCommConfigConsistency, false));
208 1 : CHK_RET(WaitActiveAsyncComplete(sockets, remoteRanks, roles, remoteExchangeInfoLens, localExchangeInfoLen, false));
209 :
210 : // 存储对端交换信息
211 2 : for (u32 i = 0; i < sockets.size(); i++) {
212 1 : if (remoteExchangeInfoLens[i] > 0 && !remoteUserDatas[i].empty()) {
213 0 : CHK_RET(collCommConfigConsistency.StoreRemoteExchangeInfo(remoteRanks[i], remoteUserDatas[i]));
214 : }
215 : }
216 :
217 1 : HCCL_INFO("[ExchangeUserInfoAsync] suc.");
218 1 : return HCCL_SUCCESS;
219 1 : }
220 :
221 : // 异步交换info数据的一个阶段（isServerRecv=true: SERVER先Recv; false: SERVER先Send）
222 5 : HcclResult ExchangeInfoMgr::ExchangeAsyncDataPhase(
223 : const std::vector<Hccl::Socket*>& sockets, const std::vector<HcommSocketRole>& roles,
224 : const std::vector<u32>& remoteRanks, std::vector<std::vector<u8>>& remoteUserDatas,
225 : const std::vector<u32>& remoteExchangeInfoLens, u32 localExchangeInfoLen,
226 : CollCommConfigConsistency& collCommConfigConsistency, bool isServerRecv) const
227 : {
228 11 : for (u32 i = 0; i < sockets.size(); i++) {
229 7 : bool shouldRecv = (roles[i] == HCOMM_SOCKET_ROLE_SERVER) == isServerRecv;
230 7 : if (shouldRecv) {
231 4 : CHK_PRT_RET(
232 : remoteExchangeInfoLens[i] > HCCL_EXCHANGE_INFO_LEN,
233 : HCCL_ERROR(
234 : "[ExchangeUserInfoAsync] remoteExchangeInfoLen[%u] for remoteRank[%u] "
235 : "exceeds max allowed [%u].",
236 : remoteExchangeInfoLens[i], remoteRanks[i], HCCL_EXCHANGE_INFO_LEN),
237 : HCCL_E_PARA);
238 3 : remoteUserDatas[i].resize(remoteExchangeInfoLens[i], 0);
239 3 : sockets[i]->RecvAsync(remoteUserDatas[i].data(), remoteExchangeInfoLens[i]);
240 : } else {
241 3 : std::vector<u8> exchangeBuf;
242 3 : collCommConfigConsistency.GetExchangeInfoBuf(exchangeBuf);
243 3 : sockets[i]->SendAsync(exchangeBuf.data(), localExchangeInfoLen);
244 3 : }
245 : }
246 4 : return HCCL_SUCCESS;
247 : }
248 :
249 : // 批量异步交换定长数据（SERVER先Recv再Send，CLIENT先Send再Recv，防死锁）
250 3 : HcclResult ExchangeInfoMgr::BatchExchangeFixedData(
251 : const std::vector<Hccl::Socket*>& sockets, const std::vector<u32>& remoteRanks,
252 : const std::vector<HcommSocketRole>& roles, const u8* sendData, u32 sendLen, u8* recvData, u32 recvLen,
253 : bool isAsync) const
254 : {
255 3 : if (!isAsync) {
256 : // 确保 Socket 连接完成
257 1 : CHK_RET(WaitAllAsyncComplete(sockets, remoteRanks));
258 : // 使用 Socket 同步收发接口
259 2 : for (u32 i = 0; i < sockets.size(); i++) {
260 1 : sockets[i]->Send(sendData, sendLen);
261 1 : sockets[i]->Recv(recvData + i * recvLen, recvLen);
262 : }
263 1 : return HCCL_SUCCESS;
264 : }
265 :
266 2 : CHK_RET(WaitAllAsyncComplete(sockets, remoteRanks));
267 : // SERVER先Recv/CLIENT先Send
268 4 : for (u32 i = 0; i < sockets.size(); i++) {
269 2 : if (roles[i] == HCOMM_SOCKET_ROLE_SERVER) {
270 0 : sockets[i]->RecvAsync(recvData + i * recvLen, recvLen);
271 : } else {
272 2 : sockets[i]->SendAsync(sendData, sendLen);
273 : }
274 : }
275 2 : CHK_RET(WaitAllAsyncComplete(sockets, remoteRanks));
276 :
277 : // SERVER再Send/CLIENT再Recv
278 4 : for (u32 i = 0; i < sockets.size(); i++) {
279 2 : if (roles[i] == HCOMM_SOCKET_ROLE_SERVER) {
280 0 : sockets[i]->SendAsync(sendData, sendLen);
281 : } else {
282 2 : sockets[i]->RecvAsync(recvData + i * recvLen, recvLen);
283 : }
284 : }
285 2 : CHK_RET(WaitAllAsyncComplete(sockets, remoteRanks));
286 :
287 2 : return HCCL_SUCCESS;
288 : }
289 :
290 9 : HcclResult ExchangeInfoMgr::WaitAllAsyncComplete(
291 : const std::vector<Hccl::Socket*>& sockets, const std::vector<u32>& remoteRanks) const
292 : {
293 9 : auto timeout = std::chrono::seconds(Hccl::EnvConfig::GetInstance().GetSocketConfig().GetLinkTimeOut());
294 9 : auto startTime = std::chrono::steady_clock::now();
295 9 : std::vector<bool> done(sockets.size(), false);
296 9 : uint32_t doneCount = 0;
297 :
298 18 : while (doneCount < sockets.size()) {
299 19 : for (size_t i = 0; i < sockets.size(); i++) {
300 10 : if (done[i]) {
301 10 : continue;
302 : }
303 10 : Hccl::SocketStatus status = sockets[i]->GetAsyncStatus();
304 10 : if (status == Hccl::SocketStatus::OK) {
305 10 : done[i] = true;
306 10 : doneCount++;
307 10 : continue;
308 : }
309 0 : if (status == Hccl::SocketStatus::TIMEOUT) {
310 0 : auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
311 0 : std::chrono::steady_clock::now() - startTime)
312 0 : .count();
313 0 : HCCL_ERROR(
314 : "[WaitAllAsyncComplete] socket timeout for remoteRank[%u], elapsed[%lld]ms.", remoteRanks[i],
315 : elapsed);
316 0 : return HCCL_E_TIMEOUT;
317 : }
318 : }
319 9 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
320 : auto elapsed
321 0 : = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - startTime)
322 0 : .count();
323 0 : for (size_t i = 0; i < sockets.size(); i++) {
324 0 : if (!done[i]) {
325 0 : HCCL_ERROR(
326 : "[WaitAllAsyncComplete] wall-clock timeout for remoteRank[%u], elapsed[%lld]ms.",
327 : remoteRanks[i], elapsed);
328 : }
329 : }
330 0 : return HCCL_E_TIMEOUT;
331 : }
332 : }
333 :
334 : auto elapsed
335 9 : = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - startTime).count();
336 9 : HCCL_INFO("[WaitAllAsyncComplete] all[%zu] sockets completed, elapsed[%lld]ms.", sockets.size(), elapsed);
337 9 : return HCCL_SUCCESS;
338 9 : }
339 :
340 : // 收集并等待有实际异步操作的socket子集
341 2 : HcclResult ExchangeInfoMgr::WaitActiveAsyncComplete(
342 : const std::vector<Hccl::Socket*>& sockets, const std::vector<u32>& remoteRanks,
343 : const std::vector<HcommSocketRole>& roles, const std::vector<u32>& remoteExchangeInfoLens, u32 localExchangeInfoLen,
344 : bool isFirstPass) const
345 : {
346 2 : std::vector<Hccl::Socket*> activeSockets;
347 2 : std::vector<u32> activeRanks;
348 4 : for (u32 i = 0; i < sockets.size(); i++) {
349 4 : bool isActive = isFirstPass ? (roles[i] == HCOMM_SOCKET_ROLE_SERVER && remoteExchangeInfoLens[i] > 0)
350 1 : || (roles[i] != HCOMM_SOCKET_ROLE_SERVER && localExchangeInfoLen > 0) :
351 1 : (roles[i] == HCOMM_SOCKET_ROLE_SERVER && localExchangeInfoLen > 0)
352 1 : || (roles[i] != HCOMM_SOCKET_ROLE_SERVER && remoteExchangeInfoLens[i] > 0);
353 2 : if (isActive) {
354 1 : activeSockets.push_back(sockets[i]);
355 1 : activeRanks.push_back(remoteRanks[i]);
356 : }
357 : }
358 2 : if (!activeSockets.empty()) {
359 1 : CHK_RET(WaitAllAsyncComplete(activeSockets, activeRanks));
360 : }
361 2 : return HCCL_SUCCESS;
362 2 : }
363 : } // namespace hccl
|
