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