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