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 "snap_shot_parse.h"
12 : #include "checkcrc.h"
13 : #include "ip_address.h"
14 : #include "comm_manager.h"
15 :
16 : using namespace Hccl;
17 :
18 100 : SnapShotParser &SnapShotParser::GetInstance()
19 : {
20 100 : static SnapShotParser instance;
21 100 : return instance;
22 : }
23 :
24 : // 保持快照，需要确保快照保持的是二进制流，需要确保换节点后能够继续使用，做save是使用，读出来dump，之后清空
25 18 : BinaryStream &SnapShotParser::GetSnapShotBuf()
26 : {
27 18 : return snapShotStream_;
28 : }
29 :
30 : // 恢复通信域，recorver调用，把传入的备份流流反序列化到本地结构
31 4 : HcclResult SnapShotParser::ParseSnapshotToLocalBuff(void *snapshotBuf, uint32_t snapshotBufSize,
32 : SnapShotBuf &localBuff)
33 : {
34 : try {
35 6 : CHK_PTR_NULL(snapshotBuf);
36 4 : if (snapshotBufSize == 0) {
37 3 : HCCL_ERROR("[%s] snapshotBufSize is 0.", __func__);
38 1 : return HcclResult::HCCL_E_PARA;
39 : }
40 :
41 3 : uint32_t dataLen = *static_cast<uint32_t *>(snapshotBuf); // 获取存储的 size
42 9 : HCCL_INFO("[%s] dataLen[%u], snapshotBufSize[%u]", __func__, dataLen, snapshotBufSize);
43 3 : if (dataLen + sizeof(dataLen) + sizeof(u32) != snapshotBufSize) {
44 3 : HCCL_ERROR("[%s] The storage size does not match the input size.", __func__);
45 1 : return HcclResult::HCCL_E_PARA;
46 : }
47 :
48 2 : char *pCur = static_cast<char *>(snapshotBuf);
49 2 : pCur = pCur + sizeof(dataLen); // 跳过头4个字节，指向crc值
50 :
51 : // 获取crc值
52 2 : u32 crcValue = *reinterpret_cast<u32 *>(pCur);
53 2 : pCur = pCur + sizeof(u32); // 跳过crc 指向真正的数据段
54 6 : HCCL_INFO("[%s] crcValue[%u]", __func__, crcValue);
55 :
56 2 : std::vector<char> bufVec(pCur, pCur + dataLen);
57 2 : BinaryStream buf(bufVec);
58 :
59 : // 校验crc
60 2 : CHK_RET(CheckBufCrc32(buf, crcValue));
61 :
62 6 : HCCL_INFO("snapshotBuf start");
63 2 : CHK_RET(DeAllSnapShotStaticBuf(buf, localBuff));
64 1 : CHK_RET(DeAllSnapShotDynamicBuf(buf, localBuff));
65 1 : DeserializeCcuStatusBuf(buf, localBuff);
66 4 : } catch (std::exception &e) {
67 3 : HCCL_ERROR("[%s]Failed, exception caught:%s, please check input snapshot!", __func__, e.what());
68 1 : return HCCL_E_INTERNAL;
69 1 : } catch (...) {
70 0 : HCCL_ERROR("parse snapshot fail, please check input snapshot!");
71 0 : return HCCL_E_INTERNAL;
72 0 : };
73 3 : HCCL_INFO("[%s] end, napshotBuf is:subGroup num[%u], comm groupName[%s]", __func__,
74 : localBuff.groupNum, localBuff.snapshot.groupName);
75 1 : return HCCL_SUCCESS;
76 : }
77 :
78 2 : HcclResult SnapShotParser::DeAllSnapShotStaticBuf(BinaryStream &buf, SnapShotBuf &localBuff)
79 : {
80 6 : HCCL_INFO("[%s] start", __func__);
81 : // 解析 静态流公共数据 SnapShotPub
82 2 : DeserializeCommVersionInfo(buf, localBuff.snapShotPub);
83 : // 解析 全局通信域
84 2 : string tmpGroupName;
85 2 : buf >> tmpGroupName;
86 2 : DeserializeCommInfo(buf, localBuff.snapshot);
87 2 : s32 ret = memcpy_s(localBuff.snapshot.groupName, sizeof(localBuff.snapshot.groupName) ,tmpGroupName.c_str(), tmpGroupName.size());
88 2 : if (ret != 0) {
89 0 : THROW<InternalException>(StringFormat("[%s] memcpy_s failed, ret=%d", __func__, ret));
90 : }
91 : // 获取 子通信域 静态buf数量
92 2 : size_t groupNum{0};
93 2 : buf >> groupNum;
94 2 : localBuff.groupNum = static_cast<uint32_t>(groupNum);
95 6 : HCCL_INFO("subGroup num[%u], groupNum[%u]", groupNum, localBuff.groupNum);
96 2 : if (localBuff.groupNum == 0) {
97 3 : HCCL_INFO("snapShot static part is empty.");
98 : } else {
99 1 : localBuff.subSnapshot.resize(localBuff.groupNum);
100 2 : for (auto& subSnapshot : localBuff.subSnapshot) {
101 : // 解析 子通信域
102 1 : string subComGroupName;
103 1 : buf >> subComGroupName;
104 1 : DeserializeSubCommInfo(buf, subSnapshot);
105 1 : s32 tmpRet = memcpy_s(subSnapshot.groupName, sizeof(subSnapshot.groupName),subComGroupName.c_str(), subComGroupName.size());
106 1 : if (tmpRet != 0) {
107 0 : THROW<InternalException>(StringFormat("[%s] memcpy_s failed, ret=%d", __func__, tmpRet));
108 : }
109 1 : }
110 : }
111 6 : HCCL_INFO("[%s] end", __func__);
112 :
113 2 : return HCCL_SUCCESS;
114 2 : }
115 :
116 1 : HcclResult SnapShotParser::DeAllSnapShotDynamicBuf(BinaryStream &buf, SnapShotBuf &localBuff)
117 : {
118 3 : HCCL_INFO("[%s] start", __func__);
119 :
120 1 : buf >> localBuff.snapShotPub.step;
121 :
122 : // 解析 全局通信域 动态短流
123 1 : SnapShotDynamic commDynamicInfo;
124 1 : CHK_RET(DeSnapShotDynamicBuf(buf, commDynamicInfo));
125 1 : localBuff.snapshot.snapShotComm.levelRankPairs = commDynamicInfo.levelRankPairs;
126 1 : localBuff.snapshot.snapShotComm.submittedOpCnt = commDynamicInfo.submittedOpCnt;
127 1 : localBuff.snapshot.snapShotComm.opMode = commDynamicInfo.opMode;
128 1 : localBuff.snapshot.snapShotComm.linkGroupPair = commDynamicInfo.linkGroupPair;
129 1 : localBuff.snapshot.snapShotComm.opExecuteConfig = commDynamicInfo.opExecuteConfig;
130 1 : localBuff.snapshot.snapShotComm.commExecuteConfig = commDynamicInfo.commExecuteConfig;
131 1 : localBuff.snapshot.snapShotComm.isLoadOp = commDynamicInfo.isLoadOp;
132 :
133 : // 获取 子通信域 动态buf 数量
134 1 : std::size_t groupNum{0};
135 1 : buf >> groupNum;
136 1 : std::size_t subSnapshotSize = localBuff.subSnapshot.size();
137 3 : HCCL_INFO("submittedOpCnt[%u] step[%u] subGroup num[%u]",
138 : localBuff.snapshot.snapShotComm.submittedOpCnt, localBuff.snapShotPub.step, groupNum);
139 1 : if (subSnapshotSize != groupNum) {
140 0 : HCCL_ERROR("[%s], subSnapshot size is wrong ,subSnapshot size is %u, "
141 : "subCommDynamicInfo Size is %u", __func__,
142 : subSnapshotSize, groupNum);
143 0 : return HcclResult::HCCL_E_INTERNAL;
144 : }
145 :
146 2 : for (auto& subSnapshot : localBuff.subSnapshot) {
147 1 : SnapShotDynamic subCommDynamicInfo;
148 1 : CHK_RET(DeSnapShotDynamicBuf(buf, subCommDynamicInfo));
149 1 : subSnapshot.snapShotSubComm.levelRankPairs = subCommDynamicInfo.levelRankPairs;
150 1 : subSnapshot.snapShotSubComm.submittedOpCnt = subCommDynamicInfo.submittedOpCnt;
151 1 : subSnapshot.snapShotSubComm.opMode = subCommDynamicInfo.opMode;
152 1 : subSnapshot.snapShotSubComm.linkGroupPair = subCommDynamicInfo.linkGroupPair;
153 1 : subSnapshot.snapShotSubComm.opExecuteConfig = subCommDynamicInfo.opExecuteConfig;
154 1 : subSnapshot.snapShotSubComm.commExecuteConfig = subCommDynamicInfo.commExecuteConfig;
155 1 : subSnapshot.snapShotSubComm.isLoadOp = subCommDynamicInfo.isLoadOp;
156 1 : }
157 3 : HCCL_INFO("[%s] end", __func__);
158 :
159 1 : return HCCL_SUCCESS;
160 1 : }
161 :
162 2 : void SnapShotParser::DeserializeCcuStatusBuf(BinaryStream &buf, SnapShotBuf &localBuff) const
163 : {
164 6 : HCCL_INFO("[%s] start", __func__);
165 :
166 2 : size_t useMsCommIdsSize{0};
167 2 : buf >> useMsCommIdsSize;
168 2 : if (useMsCommIdsSize > MAX_NUM_COMM_USING_MS) {
169 0 : THROW<InternalException>(StringFormat("[%s] useMsCommIdsSize[%zu] > MAX_NUM_COMM_USING_MS[%u]", __func__, useMsCommIdsSize, MAX_NUM_COMM_USING_MS));
170 : }
171 6 : HCCL_INFO("[SnapShotParser][%s] useMsCommIdsSize = [%u]", __func__, useMsCommIdsSize);
172 2 : localBuff.ccuStatusSnapshot.useMsCommIds.resize(useMsCommIdsSize);
173 3 : for (auto &useMsCommIdCharArr : localBuff.ccuStatusSnapshot.useMsCommIds) {
174 1 : string useMsCommId;
175 1 : buf >> useMsCommId;
176 3 : HCCL_INFO("[SnapShotParser][%s] useMsCommId is %s", __func__, useMsCommId.c_str());
177 : s32 ret
178 2 : = memcpy_s(useMsCommIdCharArr.data(), sizeof(useMsCommIdCharArr), useMsCommId.c_str(), useMsCommId.size());
179 1 : if (ret != 0) {
180 0 : THROW<InternalException>(StringFormat("[%s] memcpy_s failed, ret=%d", __func__, ret));
181 : }
182 1 : }
183 :
184 2 : size_t useSchedCommIdsSize{0};
185 2 : buf >> useSchedCommIdsSize;
186 6 : HCCL_INFO("[SnapShotParser][%s] useSchedCommIdsSize = [%u]", __func__, useSchedCommIdsSize);
187 2 : localBuff.ccuStatusSnapshot.useSchedCommIds.resize(useSchedCommIdsSize);
188 5 : for (auto &useSchedCommIdCharArr : localBuff.ccuStatusSnapshot.useSchedCommIds) {
189 3 : string useSchedCommId;
190 3 : buf >> useSchedCommId;
191 9 : HCCL_INFO("[SnapShotParser][%s] useSchedCommId is %s", __func__, useSchedCommId.c_str());
192 6 : s32 ret = memcpy_s(useSchedCommIdCharArr.data(), sizeof(useSchedCommIdCharArr), useSchedCommId.c_str(),
193 : useSchedCommId.size());
194 3 : if (ret != 0) {
195 0 : THROW<InternalException>(StringFormat("[%s] memcpy_s failed, ret=%d", __func__, ret));
196 : }
197 3 : }
198 6 : HCCL_INFO("[%s] end", __func__);
199 2 : }
200 :
201 : // 生成 单个通信域 动态短流
202 1 : HcclResult SnapShotParser::SerializeDynamicInfo(const std::vector<std::pair<u32, RankId>>& levelRankPairs,
203 : u32 submittedOpCnt, BinaryStream &binStream) const
204 : {
205 1 : size_t count = levelRankPairs.size();
206 1 : binStream << count;
207 3 : HCCL_INFO("[%s], levelRankPairs Size[%u]", __func__, count);
208 5 : for (auto &pair : levelRankPairs) {
209 4 : binStream << pair.first << pair.second;
210 : }
211 1 : binStream << submittedOpCnt;
212 3 : HCCL_INFO("submittedOpCnt[%u]", submittedOpCnt);
213 1 : return HcclResult::HCCL_SUCCESS;
214 : }
215 :
216 : // 解析 单个通信域 动态短流
217 5 : HcclResult SnapShotParser::DeSnapShotDynamicBuf(BinaryStream &buf, SnapShotDynamic &dynamicInfo) const
218 : {
219 : try {
220 5 : u32 opAccState{0};
221 5 : buf >> opAccState;
222 15 : HCCL_DEBUG("[%s], opAccState[%u]", __func__, opAccState);
223 5 : dynamicInfo.opExecuteConfig.accState = static_cast<AcceleratorState::Value>(opAccState);
224 5 : u32 commAccState{0};
225 5 : buf >> commAccState;
226 15 : HCCL_DEBUG("[%s], commAccState[%u]", __func__, commAccState);
227 5 : dynamicInfo.commExecuteConfig.accState = static_cast<AcceleratorState::Value>(commAccState);
228 5 : buf >> dynamicInfo.isLoadOp;
229 15 : HCCL_DEBUG("[%s], isLoadOp[%d]", __func__, dynamicInfo.isLoadOp);
230 :
231 5 : buf >> dynamicInfo.submittedOpCnt;
232 15 : HCCL_INFO("[%s], submittedOpCnt[%u]", __func__, dynamicInfo.submittedOpCnt);
233 5 : if (dynamicInfo.submittedOpCnt == 0) {
234 2 : return HcclResult::HCCL_SUCCESS;
235 : }
236 :
237 3 : buf >> dynamicInfo.opMode;
238 9 : HCCL_DEBUG("[%s], opMode[%d]", __func__, static_cast<s32>(dynamicInfo.opMode));
239 :
240 3 : std::size_t count{0};
241 3 : buf >> count;
242 9 : HCCL_INFO("levelRankPairs Size[%u]", count);
243 3 : if (count == 0) {
244 0 : HCCL_ERROR("[%s], levelRankPairs is empty.", __func__);
245 0 : return HcclResult::HCCL_E_INTERNAL;
246 : }
247 3 : dynamicInfo.levelRankPairs.resize(count);
248 6 : for (auto &pair : dynamicInfo.levelRankPairs) {
249 3 : buf >> pair.first >> pair.second; // 反序列化vector中的每个pair
250 : }
251 :
252 3 : size_t linkGroupPairCount{0};
253 3 : buf >> linkGroupPairCount;
254 9 : HCCL_INFO("[%s], linkGroupPairCount[%u]", __func__, linkGroupPairCount);
255 3 : dynamicInfo.linkGroupPair.resize(linkGroupPairCount);
256 5 : for (auto &linkGroupPair : dynamicInfo.linkGroupPair) {
257 2 : LinkGroup &linkGroup = linkGroupPair.first;
258 2 : u32 &cntCkeNum = linkGroupPair.second;
259 2 : size_t linkSize{0};
260 2 : buf >> linkSize;
261 4 : for (size_t i = 0; i < linkSize; ++i) {
262 : RankId rankId;
263 : u32 dieId;
264 2 : buf >> rankId >> dieId;
265 2 : IpAddress localAddr(buf);
266 2 : IpAddress remoteAddr(buf);
267 6 : HCCL_INFO("[%s], rankId[%d], dieId[%u], localAddr[%s], remoteAddr[%s]",
268 : __func__, rankId, dieId, localAddr.Describe().c_str(), remoteAddr.Describe().c_str());
269 2 : LinkInfo linkInfo{rankId, dieId, localAddr, remoteAddr};
270 2 : linkGroup.AddLink(linkInfo);
271 : }
272 2 : buf >> cntCkeNum;
273 6 : HCCL_INFO("[%s], cntCkeNum[%u], linkSize[%u]", __func__, cntCkeNum, linkSize);
274 : }
275 0 : } catch (HcclException &e) {
276 0 : HCCL_ERROR(e.what());
277 0 : return e.GetErrorCode();
278 0 : } catch (exception &e) {
279 0 : HCCL_ERROR(e.what());
280 0 : return HcclResult::HCCL_E_INTERNAL;
281 0 : } catch (...) {
282 0 : HCCL_ERROR("Unknown error occurs!");
283 0 : return HcclResult::HCCL_E_INTERNAL;
284 0 : }
285 3 : return HcclResult::HCCL_SUCCESS;
286 : }
287 : /* 全局通信域静态信息序列化 */
288 39 : void SnapShotParser::SerializeCommonInfo(const CommParams &commParams, const HcclCommConfig &config, std::unique_ptr<RankTableInfo> ranktableInfo,
289 : std::shared_ptr<TopoInfo>& topoInfo, BinaryStream &binStream) const
290 : {
291 117 : HCCL_INFO("Snapshot saving: Start to serialize static info.");
292 : // 1. 配置信息序列化
293 39 : SerializeCommConfigInfo(config, binStream);
294 : // 2. 参数信息序列化
295 39 : SerializeParamsInfo(commParams, binStream);
296 : // 3. rankTableInfo及crc信息序列化
297 39 : SerializeRankTableInfo(std::move(ranktableInfo), binStream);
298 : // 4. topoInfo及crc信息序列化
299 39 : SerializeTopoInfo(topoInfo, binStream);
300 39 : }
301 :
302 4 : void SnapShotParser::SerializeCommVersionInfo(BinaryStream &binStream) const
303 : {
304 12 : HCCL_INFO("[%s]Snapshot saving: Start to serialize comm version info.", __func__);
305 4 : char snapshotVersion[SNAPSHOT_VERSION_SIZE] = "snapshotVersionIsFixNow";
306 4 : char cannVersion[SNAPSHOT_VERSION_SIZE] = "cannVersionIsFixNow";
307 4 : char hcclVersion[SNAPSHOT_VERSION_SIZE] = "hcclVersionIsFixNow";
308 4 : binStream << snapshotVersion
309 4 : << cannVersion
310 4 : << hcclVersion;
311 4 : }
312 :
313 39 : void SnapShotParser::SerializeCommConfigInfo(const HcclCommConfig &config, BinaryStream &binStream) const
314 : {
315 117 : HCCL_INFO("[%s]Snapshot saving: Start to serial commConfig info, hcclBufferSize[%u] MB", __func__, config.hcclBufferSize);
316 39 : binStream << config.reserved << config.hcclBufferSize << config.hcclDeterministic << config.hcclCommName
317 39 : << config.hcclUdi;
318 39 : }
319 :
320 39 : void SnapShotParser::SerializeParamsInfo(const CommParams &commParams, BinaryStream &binStream) const
321 : {
322 117 : HCCL_INFO("[%s]Snapshot saving: Start to serialize commParams: commId[%s], myRank[%d], "
323 : "rankSize[%u],rankInParentComm[%d], devType[%u], devUsed[%u]", __func__,
324 : commParams.commId.c_str(), commParams.myRank, commParams.rankSize, commParams.rankInParentComm,
325 : static_cast<u32>(commParams.devType), commParams.devUsed);
326 39 : binStream << commParams.commId << commParams.myRank << commParams.rankSize << commParams.rankInParentComm
327 39 : << static_cast<u32>(commParams.devType) << commParams.devUsed;
328 39 : }
329 :
330 39 : void SnapShotParser::SerializeRankTableInfo(std::unique_ptr<RankTableInfo> ranktableInfo, BinaryStream &binStream) const
331 : {
332 117 : HCCL_INFO("[%s]Snapshot saving: Start to serialize rankTableInfo.", __func__);
333 39 : if(ranktableInfo == nullptr) {
334 99 : HCCL_WARNING("ranktableInfo is NULL.");
335 33 : return;
336 : }
337 6 : ranktableInfo->GetBinStream(true, binStream);
338 : }
339 :
340 39 : void SnapShotParser::SerializeTopoInfo(const std::shared_ptr<TopoInfo>& topoInfo, BinaryStream &binStream) const
341 : {
342 117 : HCCL_INFO("[%s]Snapshot saving: Start to serialize topoInfo.", __func__);
343 39 : if(topoInfo == nullptr) {
344 99 : HCCL_WARNING("topoInfo is NULL.");
345 33 : return;
346 : }
347 6 : topoInfo->GetBinStream(binStream);
348 : }
349 :
350 : /* 全局通信域静态信息的反序列化 */
351 1 : HcclResult SnapShotParser::DeserializeCommInfo(BinaryStream &binaryStream, Snapshot &snapShot)
352 : {
353 3 : HCCL_INFO("[%s]Snapshot recovering: Start to deserialize static info.", __func__);
354 1 : CHK_RET(DeserializeCommConfigInfo(binaryStream, snapShot.snapShotComm.config));
355 1 : CHK_RET(DeserializeParamsInfo(binaryStream, snapShot.snapShotComm.commParams));
356 1 : CHK_RET(
357 : DeserializeRankTableInfo(binaryStream, snapShot.snapShotComm.rankTableInfo));
358 1 : CHK_RET(DeserializeTopoInfo(binaryStream, snapShot.snapShotComm.topoInfo));
359 1 : return HCCL_SUCCESS;
360 : }
361 :
362 2 : HcclResult SnapShotParser::DeserializeCommVersionInfo(BinaryStream &binaryStream, SnapShotPub &snapshotPub) const
363 : {
364 6 : HCCL_INFO("[%s]Snapshot recovering: Start to deserialize commConfig info.", __func__);
365 2 : binaryStream >> snapshotPub.snapshotVersion >> snapshotPub.cannVersion >> snapshotPub.hcclVersion;
366 2 : return HCCL_SUCCESS;
367 : }
368 :
369 1 : HcclResult SnapShotParser::DeserializeCommConfigInfo(BinaryStream &binaryStream, HcclCommConfig &config) const
370 : {
371 3 : HCCL_INFO("[%s]Snapshot recovering: Start to deserialize commConfig info.", __func__);
372 1 : binaryStream >> config.reserved >> config.hcclBufferSize >> config.hcclDeterministic >> config.hcclCommName
373 1 : >> config.hcclUdi;
374 1 : config.hcclBufferSize = 0;
375 3 : HCCL_INFO("Snapshot recovering: hcclBufferSize[%u] MB, hcclDeterministic[%u], hcclCommName[%s], "
376 : "hcclUdi[%s]",
377 : config.hcclBufferSize, config.hcclDeterministic, config.hcclCommName, config.hcclUdi);
378 1 : return HCCL_SUCCESS;
379 : }
380 :
381 1 : HcclResult SnapShotParser::DeserializeParamsInfo(BinaryStream& binaryStream, Hccl::CommParams& commParams) const
382 : {
383 3 : HCCL_INFO("[%s]Snapshot recovering: Start to deserialize params info.", __func__);
384 1 : binaryStream >> commParams.commId;
385 1 : binaryStream >> commParams.myRank;
386 1 : binaryStream >> commParams.rankSize;
387 1 : binaryStream >> commParams.rankInParentComm;
388 1 : u32 dev = 0;
389 1 : binaryStream >> dev;
390 1 : commParams.devType = static_cast<DevType::Value>(dev);
391 1 : binaryStream >> commParams.devUsed;
392 1 : commParams.isWorldGroup = true;
393 3 : HCCL_INFO("Snapshot recovering: commId[%s], myRank[%d], rankSize[%u],rankInParentComm[%d], devType[%u], devUsed[%u]"
394 : , commParams.commId.c_str(), commParams.myRank, commParams.rankSize, commParams.rankInParentComm,
395 : static_cast<u32>(commParams.devType), commParams.devUsed);
396 1 : return HCCL_SUCCESS;
397 : }
398 :
399 1 : HcclResult SnapShotParser::DeserializeRankTableInfo(BinaryStream& binaryStream, RankTableInfo& rankTableInfo) const
400 : {
401 3 : HCCL_INFO("[%s]Snapshot recovering: Start to dserialized rankTable info.", __func__);
402 1 : rankTableInfo = RankTableInfo(binaryStream);
403 1 : return HCCL_SUCCESS;
404 : }
405 :
406 1 : HcclResult SnapShotParser::DeserializeTopoInfo(BinaryStream& binaryStream, TopoInfo& topoInfo) const
407 : {
408 3 : HCCL_INFO("[%s]Snapshot recovering: Start to dserialized topo info.", __func__);
409 1 : topoInfo = TopoInfo(binaryStream);
410 1 : return HCCL_SUCCESS;
411 : }
412 : /* 子通信域静态信息的序列化 */
413 2 : void SnapShotParser::SerializeSubCommInfo(const CommParams &commParams, const HcclCommConfig &subConfig,
414 : const std::vector<u32> &rankId, BinaryStream &binStream) const
415 : {
416 6 : HCCL_INFO("[%s]Snapshot saving: Start to serial subComm static info.", __func__);
417 2 : SerializeSubCommParamsInfo(commParams, binStream);
418 2 : SerializeSubCommConfigInfo(subConfig, binStream);
419 2 : SerializeRankIds(rankId, binStream);
420 2 : }
421 :
422 2 : void SnapShotParser::SerializeSubCommParamsInfo(const CommParams &commParams,BinaryStream &binStream) const
423 : {
424 6 : HCCL_INFO("[%s]Snapshot saving: Start to serialize sub commParams: commId[%s], myRank[%d], "
425 : "rankSize[%u],rankInParentComm[%d], devType[%u], devUsed[%u]",
426 : __func__, commParams.commId.c_str(), commParams.myRank, commParams.rankSize, commParams.rankInParentComm,
427 : static_cast<u32>(commParams.devType), commParams.devUsed);
428 2 : binStream << commParams.commId << commParams.myRank << commParams.rankSize << commParams.rankInParentComm
429 2 : << static_cast<u32>(commParams.devType) << commParams.devUsed;
430 2 : }
431 :
432 2 : void SnapShotParser::SerializeSubCommConfigInfo(const HcclCommConfig &subConfig, BinaryStream &binStream) const
433 : {
434 6 : HCCL_INFO("Snapshot recovering: hcclBufferSize[%u] MB, hcclDeterministic[%u], hcclCommName[%s], hcclUdi[%s]",
435 : subConfig.hcclBufferSize, subConfig.hcclDeterministic, subConfig.hcclCommName, subConfig.hcclUdi);
436 2 : binStream << subConfig.reserved << subConfig.hcclBufferSize << subConfig.hcclDeterministic << subConfig.hcclCommName
437 2 : << subConfig.hcclUdi;
438 2 : }
439 :
440 2 : void SnapShotParser::SerializeRankIds(const std::vector<u32> &rankIds, BinaryStream &binStream) const
441 : {
442 6 : HCCL_INFO("[%s]Snapshot saving: Start to serialize rankIds.", __func__);
443 2 : binStream << rankIds.size();
444 6 : HCCL_INFO("rankIdsSize[%u]", rankIds.size());
445 10 : for (auto rankId : rankIds) {
446 8 : binStream << rankId;
447 : }
448 2 : }
449 :
450 : /* 子通信域静态信息的反序列化 */
451 2 : HcclResult SnapShotParser::DeserializeSubCommInfo(BinaryStream& stream, SubSnapshot& subSnapShot)
452 : {
453 : // 参数信息反序列化
454 2 : CHK_RET(DeserializeSubCommParamsInfo(stream, subSnapShot.snapShotSubComm.commParams));
455 : // 配置信息反序列化
456 2 : CHK_RET(DeserializeSubCommConfigInfo(stream, subSnapShot.snapShotSubComm.config));
457 : // rankIds反序列化
458 2 : CHK_RET(DeserializeRankIds(stream, subSnapShot.snapShotSubComm.rankIds));
459 2 : return HCCL_SUCCESS;
460 : }
461 :
462 2 : HcclResult SnapShotParser::DeserializeSubCommConfigInfo(BinaryStream& binaryStream, HcclCommConfig& subConfig) const
463 : {
464 6 : HCCL_INFO("[%s]Snapshot recovering: Start to deserial sub commConfig info.", __func__);
465 2 : binaryStream >> subConfig.reserved >> subConfig.hcclBufferSize >> subConfig.hcclDeterministic
466 2 : >> subConfig.hcclCommName >> subConfig.hcclUdi;
467 6 : HCCL_INFO("Snapshot recovering: hcclReserved[%s], hcclBufferSize[%u] MB, hcclDeterministic[%u], hcclCommName[%s], "
468 : "hcclUdi[%s]", subConfig.reserved, subConfig.hcclBufferSize, subConfig.hcclDeterministic,
469 : subConfig.hcclCommName, subConfig.hcclUdi);
470 2 : return HCCL_SUCCESS;
471 : }
472 :
473 2 : HcclResult SnapShotParser::DeserializeSubCommParamsInfo(BinaryStream &binaryStream, Hccl::CommParams &subCommParam) const
474 : {
475 2 : u32 dev = 0;
476 6 : HCCL_INFO("[%s]Snapshot recovering: Start to deserialize sub params info.", __func__);
477 2 : binaryStream >> subCommParam.commId >> subCommParam.myRank >> subCommParam.rankSize >> subCommParam.rankInParentComm
478 2 : >> dev >> subCommParam.devUsed;
479 2 : subCommParam.devType = static_cast<DevType::Value>(dev);
480 6 : HCCL_INFO(
481 : "Snapshot recovering: commId[%s], myRank[%d], rankSize[%u],rankInParentComm[%d], devType[%u], devUsed[%u]"
482 : , subCommParam.commId.c_str(), subCommParam.myRank, subCommParam.rankSize,
483 : subCommParam.rankInParentComm, static_cast<u32>(subCommParam.devType), subCommParam.devUsed);
484 2 : return HCCL_SUCCESS;
485 : }
486 :
487 2 : HcclResult SnapShotParser::DeserializeRankIds(BinaryStream& binaryStream, vector<RankId>& rankIds) const
488 : {
489 6 : HCCL_INFO("[%s]Snapshot recovering: Start to deserialize rankIds.", __func__);
490 : size_t rankIdsSize;
491 2 : binaryStream >> rankIdsSize;
492 6 : HCCL_INFO("rankIdsSize[%u]", rankIdsSize);
493 2 : rankIds.resize(rankIdsSize);
494 7 : for (auto &id : rankIds) {
495 5 : binaryStream >> id;
496 : }
497 2 : return HCCL_SUCCESS;
498 : }
499 :
500 1 : HcclResult SnapShotParser::CalcBufCrc32(BinaryStream& buf,u32 &crcValue) const
501 : {
502 1 : CheckCrc crc;
503 1 : auto ret = crc.Calc32Crc(buf.GetString().c_str(), buf.GetSize(), &crcValue);
504 1 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[CalcCrc32]calculate crc failed, ret[%d]", ret), ret);
505 1 : return HCCL_SUCCESS;
506 1 : }
507 :
508 3 : HcclResult SnapShotParser::CheckBufCrc32(BinaryStream& buf,const u32 otherCrcValue) const
509 : {
510 3 : CheckCrc crc;
511 3 : u32 myCrcValue = 0;
512 3 : auto ret = crc.Calc32Crc(buf.GetString().c_str(), buf.GetSize(), &myCrcValue);
513 9 : HCCL_INFO("[CheckCrc32]myCrcValue[%u] otherCrcValue[%u]", myCrcValue, otherCrcValue);
514 3 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[CheckCrc32]calculate crc failed, ret[%d]", ret), ret);
515 3 : CHK_PRT_RET(myCrcValue != otherCrcValue, HCCL_ERROR("[CalcCrc32]check crc failed, ret[%d]", ret), HCCL_E_INTERNAL);
516 3 : return HCCL_SUCCESS;
517 3 : }
518 :
519 24 : void SnapShotParser::SetIsNeedLoadOp(bool status)
520 : {
521 24 : isNeedLoadOp = status;
522 24 : }
523 1 : bool SnapShotParser::GetIsNeedLoadOp() const
524 : {
525 1 : return isNeedLoadOp;
526 : }
|
