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