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 "op_base_v2.h"
12 : #include <algorithm>
13 : #include <future>
14 : #include <map>
15 : #include <mutex>
16 : #include <fstream>
17 : #include <string>
18 : #include <hccl/hccl_types.h>
19 : #include <adapter_error_manager_pub.h>
20 :
21 : #include "hccl/base.h"
22 : #include "mc2_type.h"
23 : #include "param_check_v2.h"
24 : #include "orion_adapter_rts.h"
25 : #include "hccl_communicator.h"
26 : #include "hccl_common_v2.h"
27 : #include "log.h"
28 : #include "sal.h"
29 : #include "communicator_callback.h"
30 : #include "root_handle_v2.h"
31 : #include "rank_info_detect.h"
32 : #include "env_config.h"
33 : #include "stream_utils.h"
34 :
35 :
36 : #include "hostdpu/dpu_kernel_entrance.h"
37 : #include "hostdpu/flush_manager.h"
38 :
39 : using namespace std;
40 : using namespace Hccl;
41 : std::map<std::string, Hccl::HcclCommunicator *> g_hcclCommunicators[MAX_MODULE_DEVICE_NUM + 1];
42 :
43 1 : static OpType GetOpTypeV2(std::string opTypeName) {
44 3 : HCCL_INFO("GetOpTypeV2 start, opTypeName %s", opTypeName.c_str());
45 1 : std::transform(opTypeName.begin(), opTypeName.end(), opTypeName.begin(), ::toupper);
46 1 : auto iter = HCOM_OP_TYPE_STR_MAP_V2.find(opTypeName);
47 1 : if (iter != HCOM_OP_TYPE_STR_MAP_V2.end()) {
48 1 : return iter->second;
49 : }
50 0 : return OpType::OPTYPEINVALID;
51 : }
52 :
53 8 : static void LoadConfigCommName(string &commId, const HcclCommConfig &config)
54 : {
55 8 : if (config.hcclCommName[0] == '\0') {
56 0 : HCCL_WARNING("[LoadConfigCommName] config.hcclCommName is empty, use default commId[%s]", commId.c_str());
57 0 : return;
58 : }
59 :
60 8 : auto commNameLength = strlen(config.hcclCommName);
61 8 : commNameLength = commNameLength < COMM_NAME_MAX_LENGTH ? commNameLength : COMM_NAME_MAX_LENGTH;
62 8 : commId = std::string(config.hcclCommName, commNameLength);
63 24 : HCCL_RUN_INFO("Entry-%s: set commName[%s]", __func__, commId.c_str());
64 : }
65 :
66 :
67 : thread_local s32 g_hcclDeviceId = INVALID_INT;
68 38 : static HcclResult HcclGetDeviceId(void)
69 : {
70 38 : if (g_hcclDeviceId == INVALID_INT) {
71 3 : aclError ret = aclrtGetDevice(&g_hcclDeviceId);
72 3 : CHK_PRT_RET(ret != ACL_SUCCESS, HCCL_WARNING("[HcclGetDeviceId]aclrtGetDevice failed, ret[%d]", ret),
73 : HCCL_E_INTERNAL);
74 : }
75 38 : CHK_PRT_RET(static_cast<u32>(g_hcclDeviceId) >= MAX_MODULE_DEVICE_NUM,
76 : HCCL_WARNING("[HcclGetDeviceId]deviceLogicId[%d] is bigger than HCCL_AISERVER_DEVICE_NUM_MAX:[%u]",
77 : g_hcclDeviceId, MAX_MODULE_DEVICE_NUM), HCCL_E_INTERNAL);
78 114 : HCCL_INFO("[HcclGetDeviceId] deviceLogicId[%d] ", g_hcclDeviceId);
79 38 : return HCCL_SUCCESS;
80 : }
81 :
82 38 : static s32 HcclGetThreadDeviceId()
83 : {
84 38 : CHK_PRT_RET(HcclGetDeviceId() != HCCL_SUCCESS, HCCL_WARNING("[HcclGetThreadDeviceId] get fail deviceLogicId[%d]",
85 : g_hcclDeviceId), INVALID_INT);
86 38 : return g_hcclDeviceId;
87 : }
88 :
89 : template <typename Func>
90 2 : HcclResult HcclCommOperationImplV2(HcclComm comm, const std::string func_name, Func operate)
91 : {
92 2 : s32 deviceLogicId = HcclGetThreadDeviceId();
93 2 : s32 devPhyId = HrtGetDevicePhyIdByIndex(deviceLogicId);
94 2 : HcclUs startut = TIME_NOW();
95 2 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
96 2 : auto ret = operate(*communicator);
97 2 : CHK_PRT_RET(ret != HCCL_SUCCESS,
98 : HCCL_ERROR("[%s] errNo[0x%016llx] deviceLogicId[%d], comm[%s]", func_name.c_str(),
99 : ret, deviceLogicId, communicator->GetId().c_str()), HCCL_E_INTERNAL);
100 6 : HCCL_RUN_INFO("%s success, take time [%lld]us, deviceLogicId[%d], devPhyId[%d], comm[%s]", func_name.c_str(),
101 : DURATION_US(TIME_NOW() - startut), deviceLogicId, devPhyId, communicator->GetId().c_str());
102 2 : return HCCL_SUCCESS;
103 : }
104 :
105 : namespace {
106 : std::map<HcclReduceOp, ReduceOp> HCCL_OP_REDUCE_MAP = {{HCCL_REDUCE_SUM, ReduceOp::SUM},
107 : {HCCL_REDUCE_PROD, ReduceOp::PROD},
108 : {HCCL_REDUCE_MAX, ReduceOp::MAX},
109 : {HCCL_REDUCE_MIN, ReduceOp::MIN}};
110 : }
111 :
112 10 : static void CheckHcclDeterministic(uint32_t hcclDeterministic)
113 : {
114 10 : if (hcclDeterministic == 0) {
115 0 : HCCL_WARNING("[HcclCommInitClusterInfoConfig] hcclDeterministic[%u] is not support.", hcclDeterministic);
116 10 : } else if (hcclDeterministic != HCCL_COMM_DETERMINISTIC_CONFIG_NOT_SET && hcclDeterministic != 1) {
117 0 : HCCL_WARNING("[HcclCommInitClusterInfoConfig] hcclDeterministic[%u] is invalid.", hcclDeterministic);
118 : }
119 10 : }
120 :
121 6 : HcclResult CreateCommConfig(uint32_t rank, HcclCommConfig *config, HcclComm *comm, std::string &ranktableM)
122 : {
123 6 : string commId(HCCL_WORLD_GROUP);
124 6 : LoadConfigCommName(commId, *config);
125 :
126 6 : HcclCommInfoV2 &opbasedCommInfoV2 = GetCommInfoV2();
127 6 : CHK_PRT_RET(opbasedCommInfoV2.hcclGroupMap.find(commId) != opbasedCommInfoV2.hcclGroupMap.end(),
128 : HCCL_ERROR("[CreateCommConfig] errNo[0x%016llx] The comm name[%s] already exists in Group2Comm map.",
129 : HCCL_ERROR_CODE(HCCL_E_PARA), commId.c_str()), HCCL_E_PARA);
130 :
131 6 : bool devUsed = false;
132 6 : bool isWorldGroup = true;
133 : Hccl::CommParams commParams{commId, static_cast<Hccl::RankId>(rank), 0,
134 6 : static_cast<Hccl::RankId>(rank), Hccl::HrtGetDeviceType(), devUsed, isWorldGroup};
135 :
136 6 : shared_ptr<HcclCommConfig> hcclConf;
137 6 : EXCEPTION_CATCH((hcclConf = make_shared<HcclCommConfig>()), return HCCL_E_PTR);
138 6 : CHK_SAFETY_FUNC_RET(memcpy_s(hcclConf->reserved, sizeof(hcclConf->reserved), config->reserved, sizeof(config->reserved)));
139 :
140 6 : hcclConf->hcclBufferSize = config->hcclBufferSize;
141 6 : if (config->hcclBufferSize == HCCL_COMM_BUFFSIZE_CONFIG_NOT_SET) {
142 1 : hcclConf->hcclBufferSize = 0;
143 3 : HCCL_INFO("[HcclCommInitClusterInfoConfig] set default HCCL BUFFER");
144 : }
145 6 : CheckHcclDeterministic(config->hcclDeterministic);
146 : // 默认全都开启确定性计算
147 6 : hcclConf->hcclDeterministic = 1;
148 :
149 6 : CHK_SAFETY_FUNC_RET(memcpy_s(hcclConf->hcclCommName, sizeof(hcclConf->hcclCommName), config->hcclCommName, sizeof(config->hcclCommName)));
150 6 : CHK_SAFETY_FUNC_RET(memcpy_s(hcclConf->hcclUdi, sizeof(hcclConf->hcclUdi), config->hcclUdi, sizeof(config->hcclUdi)));
151 :
152 6 : opbasedCommInfoV2.pComm.reset(new (std::nothrow) Hccl::HcclCommunicator(commParams, hcclConf.get()));
153 6 : CHK_SMART_PTR_NULL(opbasedCommInfoV2.pComm);
154 :
155 : /* --------------初始化------------------------- */
156 6 : HcclResult ret = HCCL_SUCCESS;
157 6 : bool errorFlag = false;
158 : do {
159 6 : ret = opbasedCommInfoV2.pComm->Init(ranktableM);
160 12 : CHK_PRT_BREAK(ret != HcclResult::HCCL_SUCCESS,
161 : HCCL_ERROR("[CreateCommConfig]opbasedCommInfoV2.pComm->Init failed, errNo[0x%016llx]", HCCL_ERROR_CODE(ret)),
162 : errorFlag = true);
163 4 : opbasedCommInfoV2.pComm->RegisterAcceStateCallBack(CommunicatorCallback());
164 4 : s32 logicDevId = HrtGetDevice();
165 4 : ret = CommManager::GetInstance(logicDevId).SetCommAcceleratorV2(opbasedCommInfoV2.pComm.get(), config->hcclOpExpansionMode); // 通信域创建，设置默认accelerator
166 4 : CHK_PRT_BREAK(ret != HcclResult::HCCL_SUCCESS,
167 : HCCL_ERROR("[CreateCommConfig]SetCommAcceleratorV2 failed, errNo[0x%016llx]", HCCL_ERROR_CODE(ret)),
168 : errorFlag = true);
169 4 : *comm = static_cast<HcclComm>(opbasedCommInfoV2.pComm.get());
170 4 : HcclGetRankSizeV2(*comm, &commParams.rankSize);
171 4 : opbasedCommInfoV2.commParams = commParams;
172 :
173 4 : HcclGroupParamsV2 params{};
174 4 : params.pComm = opbasedCommInfoV2.pComm;
175 4 : params.groupRank = static_cast<Hccl::RankId>(rank);
176 4 : std::unique_lock<std::mutex> lock(opbasedCommInfoV2.groupParamsLock);
177 4 : opbasedCommInfoV2.hcclGroupMap[commId] = params;
178 :
179 8 : opbasedCommInfoV2.pComm->RegisterPrintChannelInfoCallback(
180 8 : CommManager::GetInstance(logicDevId).GetPrintChannelInfoCallback());
181 4 : } while (0);
182 :
183 6 : if (errorFlag) {
184 6 : HCCL_ERROR("[Init][CreateCommConfig]CreateCommConfig failed, rank[%u], commId[%s],"\
185 : "return[0x%016llx]", rank, commId.c_str(), HCCL_ERROR_CODE(ret));
186 2 : (void)HcclCommDestroyV2(opbasedCommInfoV2.pComm.get());
187 2 : *comm = nullptr;
188 2 : return ret;
189 : }
190 :
191 4 : return HCCL_SUCCESS;
192 6 : }
193 :
194 2 : HcclResult CreateCommConfigRootInfo(uint32_t rank, const HcclCommConfig *config, const std::string &identifier,
195 : const RankTableInfo &ranktable, HcclComm *comm)
196 : {
197 : // check
198 2 : HcclCommInfoV2 &opbasedCommInfoV2 = GetCommInfoV2();
199 2 : CHK_PRT_RET(opbasedCommInfoV2.hcclGroupMap.find(identifier) != opbasedCommInfoV2.hcclGroupMap.end(),
200 : HCCL_ERROR("[HcclCommInitRootInfoConfigV2]errNo[0x%016llx] The comm name[%s] already exists in Group2Comm map.",
201 : HCCL_ERROR_CODE(HCCL_E_PARA), identifier.c_str()), HCCL_E_PARA);
202 :
203 2 : bool devUsed = false;
204 2 : bool isWorldGroup = true;
205 : Hccl::CommParams commParams{identifier, static_cast<Hccl::RankId>(rank), 0,
206 2 : static_cast<Hccl::RankId>(rank), Hccl::HrtGetDeviceType(), devUsed, isWorldGroup};
207 :
208 2 : shared_ptr<HcclCommConfig> hcclConf;
209 2 : EXCEPTION_CATCH((hcclConf = make_shared<HcclCommConfig>()), return HCCL_E_PTR);
210 2 : CHK_SAFETY_FUNC_RET(memcpy_s(hcclConf->reserved, sizeof(hcclConf->reserved), config->reserved, sizeof(config->reserved)));
211 :
212 2 : hcclConf->hcclBufferSize = config->hcclBufferSize;
213 2 : if (config->hcclBufferSize == HCCL_COMM_BUFFSIZE_CONFIG_NOT_SET) {
214 1 : hcclConf->hcclBufferSize = 0;
215 3 : HCCL_INFO("[HcclCommInitRootInfoConfigV2] set default HCCL BUFFER");
216 : }
217 2 : CheckHcclDeterministic(config->hcclDeterministic);
218 : // 默认全都开启确定性计算
219 2 : hcclConf->hcclDeterministic = 1;
220 :
221 2 : CHK_SAFETY_FUNC_RET(memcpy_s(hcclConf->hcclCommName, sizeof(hcclConf->hcclCommName), config->hcclCommName, sizeof(config->hcclCommName)));
222 2 : CHK_SAFETY_FUNC_RET(memcpy_s(hcclConf->hcclUdi, sizeof(hcclConf->hcclUdi), config->hcclUdi, sizeof(config->hcclUdi)));
223 :
224 2 : opbasedCommInfoV2.pComm.reset(new (std::nothrow) Hccl::HcclCommunicator(commParams, hcclConf.get()));
225 2 : CHK_SMART_PTR_NULL(opbasedCommInfoV2.pComm);
226 2 : HcclResult ret = HCCL_SUCCESS;
227 2 : bool errorFlag = false;
228 : do {
229 2 : ret = opbasedCommInfoV2.pComm->Init(ranktable);
230 2 : CHK_PRT_BREAK(ret != HcclResult::HCCL_SUCCESS,
231 : HCCL_ERROR("[%s]opbasedCommInfoV2.pComm->Init failed, errNo[0x%016llx]", __func__, HCCL_ERROR_CODE(ret)),
232 : errorFlag = true);
233 2 : opbasedCommInfoV2.pComm->RegisterAcceStateCallBack(CommunicatorCallback());
234 2 : s32 logicDevId = HrtGetDevice();
235 2 : ret = CommManager::GetInstance(logicDevId).SetCommAcceleratorV2(opbasedCommInfoV2.pComm.get(), config->hcclOpExpansionMode); // 通信域创建，设置默认accelerator
236 2 : CHK_PRT_BREAK(ret != HcclResult::HCCL_SUCCESS,
237 : HCCL_ERROR("[%s]SetCommAcceleratorV2 failed, errNo[0x%016llx]", __func__, HCCL_ERROR_CODE(ret)),
238 : errorFlag = true);
239 2 : *comm = static_cast<HcclComm>(opbasedCommInfoV2.pComm.get());
240 :
241 2 : HcclGetRankSizeV2(*comm, &commParams.rankSize);
242 2 : opbasedCommInfoV2.commParams = commParams;
243 :
244 2 : HcclGroupParamsV2 params{};
245 2 : params.pComm = opbasedCommInfoV2.pComm;
246 2 : params.groupRank = static_cast<Hccl::RankId>(rank);
247 2 : std::unique_lock<std::mutex> lock(opbasedCommInfoV2.groupParamsLock);
248 2 : opbasedCommInfoV2.hcclGroupMap[identifier] = params;
249 :
250 4 : opbasedCommInfoV2.pComm->RegisterPrintChannelInfoCallback(
251 4 : CommManager::GetInstance(logicDevId).GetPrintChannelInfoCallback());
252 2 : } while (0);
253 :
254 2 : if (errorFlag) {
255 0 : HCCL_ERROR("[Init][%s]CreateCommConfigRootInfo failed return[0x%016llx] rank[%u] identifier[%s] commName[%s] udi[%s]",
256 : __func__, HCCL_ERROR_CODE(ret), rank, identifier.c_str(), hcclConf->hcclCommName, hcclConf->hcclUdi);
257 0 : (void)HcclCommDestroyV2(opbasedCommInfoV2.pComm.get());
258 0 : *comm = nullptr;
259 0 : return ret;
260 : }
261 :
262 2 : return HCCL_SUCCESS;
263 2 : }
264 :
265 : // 单算子 创建全局通信域 Json V2
266 6 : static HcclResult ParseJsonAndCreateComm(nlohmann::json& data, uint32_t rank, HcclCommConfig *config, HcclComm *comm,
267 : std::string &ranktableM)
268 : {
269 6 : CHK_PRT_RET(data.find("rank_count") == data.end(),
270 : HCCL_ERROR("[HcclCommInitClusterInfo] json object has no property called 'rank_count'"),
271 : HCCL_E_PARA);
272 :
273 6 : HcclResult ret = CreateCommConfig(rank, config, comm, ranktableM);
274 12 : CHK_PRT_RET(ret != HCCL_SUCCESS,
275 : HCCL_ERROR("[ParseJsonAndCreateComm]CreateCommConfig fail, errNo[%d]", ret), ret);
276 :
277 4 : return HCCL_SUCCESS;
278 : }
279 :
280 : // 单算子 创建全局通信域 RankTable V2
281 3 : HcclResult HcclCommInitClusterInfoV2(const char *clusterInfo, uint32_t rank, HcclComm *comm)
282 : {
283 3 : HcclUs startut = TIME_NOW();
284 3 : s32 deviceLogicId = HcclGetThreadDeviceId();
285 3 : s32 devPhyId = HrtGetDevicePhyIdByIndex(deviceLogicId);
286 3 : CHK_PTR_NULL(clusterInfo);
287 3 : CHK_PTR_NULL(comm);
288 :
289 : // 读取ranktable文件
290 3 : std::string ranktableM;
291 6 : CHK_RET(HcomLoadRankTableFileV2(clusterInfo, ranktableM));
292 :
293 2 : nlohmann::json data;
294 : try {
295 2 : data = nlohmann::json::parse(ranktableM);
296 0 : } catch (const nlohmann::json::parse_error &e) {
297 0 : HCCL_ERROR("JSON parse error: %s at byte %d", e.what(), e.byte);
298 0 : return HCCL_E_PARA;
299 0 : } catch (const nlohmann::json::exception &e) {
300 0 : HCCL_ERROR("JSON parse error: %s", e.what());
301 0 : return HCCL_E_PARA;
302 0 : } catch (...) {
303 0 : HCCL_ERROR("load allocated resource to json fail, please check json input");
304 0 : return HCCL_E_INTERNAL;
305 0 : };
306 :
307 2 : CHK_RET(CallSingletons()); // 临时规避，在初始化通信域前声明单例保证时序
308 : // check
309 2 : string commId(HCCL_WORLD_GROUP);
310 2 : HcclCommInfoV2 &opbasedCommInfoV2 = GetCommInfoV2();
311 2 : CHK_PRT_RET(opbasedCommInfoV2.hcclGroupMap.find(commId) != opbasedCommInfoV2.hcclGroupMap.end(),
312 : HCCL_ERROR("[HcclCommInitClusterInfoV2]errNo[0x%016llx] The comm name[%s] already exists in Group2Comm map.",
313 : HCCL_ERROR_CODE(HCCL_E_PARA), commId.c_str()), HCCL_E_PARA);
314 :
315 2 : bool devUsed = false;
316 2 : bool isWorldGroup = true;
317 : Hccl::CommParams commParams{commId, static_cast<Hccl::RankId>(rank), 0,
318 2 : static_cast<Hccl::RankId>(rank), Hccl::HrtGetDeviceType(), devUsed, isWorldGroup};
319 2 : opbasedCommInfoV2.pComm.reset(new (std::nothrow) Hccl::HcclCommunicator(commParams));
320 2 : CHK_PTR_NULL(opbasedCommInfoV2.pComm);
321 : /* --------------初始化------------------------- */
322 2 : HcclResult ret = HCCL_SUCCESS;
323 2 : bool errorFlag = false;
324 : do {
325 2 : ret = opbasedCommInfoV2.pComm->Init(ranktableM);
326 2 : CHK_PRT_BREAK(ret != HcclResult::HCCL_SUCCESS,
327 : HCCL_ERROR("[%s]opbasedCommInfoV2.pComm->Init failed, errNo[0x%016llx]", __func__, HCCL_ERROR_CODE(ret)),
328 : errorFlag = true);
329 2 : opbasedCommInfoV2.pComm->RegisterAcceStateCallBack(CommunicatorCallback());
330 2 : s32 logicDevId = HrtGetDevice();
331 2 : ret = CommManager::GetInstance(logicDevId).SetCommAcceleratorV2(opbasedCommInfoV2.pComm.get(), 0); // 通信域创建，设置默认accelerator
332 2 : CHK_PRT_BREAK(ret != HcclResult::HCCL_SUCCESS,
333 : HCCL_ERROR("[%s]SetCommAcceleratorV2 failed, errNo[0x%016llx]", __func__, HCCL_ERROR_CODE(ret)),
334 : errorFlag = true);
335 2 : *comm = static_cast<HcclComm>(opbasedCommInfoV2.pComm.get());
336 :
337 2 : HcclGetRankSizeV2(*comm, &commParams.rankSize);
338 2 : opbasedCommInfoV2.commParams = commParams;
339 :
340 2 : HcclGroupParamsV2 params{};
341 2 : params.pComm = opbasedCommInfoV2.pComm;
342 2 : params.groupRank = static_cast<Hccl::RankId>(rank);
343 2 : std::unique_lock<std::mutex> lock(opbasedCommInfoV2.groupParamsLock);
344 2 : opbasedCommInfoV2.hcclGroupMap[commId] = params;
345 :
346 4 : opbasedCommInfoV2.pComm->RegisterPrintChannelInfoCallback(
347 4 : CommManager::GetInstance(logicDevId).GetPrintChannelInfoCallback());
348 2 : } while (0);
349 :
350 2 : if (errorFlag) {
351 0 : HCCL_ERROR("[Init][%s]HcclCommInitClusterInfoV2 failed, clusterInfo[%s], rank[%u], deviceLogicId[%d], devPhyId[%d], commId[%s]"\
352 : "return[0x%016llx]", __func__, clusterInfo, rank,
353 : deviceLogicId, devPhyId, commId.c_str(), HCCL_ERROR_CODE(ret));
354 0 : (void)HcclCommDestroyV2(opbasedCommInfoV2.pComm.get());
355 0 : *comm = nullptr;
356 0 : return ret;
357 : }
358 : /* 关键状态记录 */
359 6 : HCCL_RUN_INFO("[HCCL_TRACE]%s success, take time [%lld]us, clusterInfo[%s], rank[%u], deviceLogicId[%d], devPhyId[%d], commId[%s].",
360 : __func__, DURATION_US(TIME_NOW() - startut), clusterInfo, rank, deviceLogicId, devPhyId, commId.c_str());
361 2 : return HCCL_SUCCESS;
362 3 : }
363 :
364 4 : HcclResult HcclCommInitClusterInfoMemConfigV2(const char *rankTableString, uint32_t rank,
365 : HcclCommConfig *config, HcclComm *comm)
366 : {
367 4 : CHK_RET(CallSingletons()); // 临时规避，在初始化通信域前声明单例保证时序
368 :
369 12 : HCCL_INFO("HcclCommInitClusterInfoMemConfigV2 Begin, commName[%s]", config->hcclCommName);
370 4 : std::string rankTableM(rankTableString);
371 4 : nlohmann::json data;
372 : try {
373 5 : data = nlohmann::json::parse(rankTableM);
374 1 : } catch (const nlohmann::json::parse_error &e) {
375 3 : HCCL_ERROR("[RankTable]JSON parse error: %s at byte %d", e.what(), e.byte);
376 1 : return HCCL_E_INTERNAL;
377 1 : } catch (const nlohmann::json::exception &e) {
378 0 : HCCL_ERROR("[RankTable]JSON parse error: %s", e.what());
379 0 : return HCCL_E_INTERNAL;
380 0 : } catch (...) {
381 0 : HCCL_ERROR("[RankTable]load allocated resource to json fail, please check json input");
382 0 : return HCCL_E_INTERNAL;
383 0 : };
384 3 : HcclResult ret = ParseJsonAndCreateComm(data, rank, config, comm, rankTableM);
385 9 : CHK_PRT_RET(ret, HCCL_ERROR("[Parse][Json]errNo[0x%016llx] and create comm failed, commName[%s.",
386 : HCCL_ERROR_CODE(ret), config->hcclCommName), static_cast<HcclResult>(ret));
387 3 : HCCL_INFO("HcclCommInitClusterInfoMemConfigV2 End, commName[%s]", config->hcclCommName);
388 1 : return HCCL_SUCCESS;
389 4 : }
390 :
391 4 : HcclResult HcclCommInitClusterInfoConfigV2(
392 : const char *clusterInfo, uint32_t rank, HcclCommConfig *config, HcclComm *comm)
393 : {
394 4 : HcclUs startut = TIME_NOW();
395 4 : s32 deviceLogicId = HcclGetThreadDeviceId();
396 4 : s32 devPhyId = HrtGetDevicePhyIdByIndex(deviceLogicId);
397 12 : HCCL_RUN_INFO("Entry-HcclCommInitClusterInfoConfig V950, commEngine[%u], commId[%s]", config->hcclOpExpansionMode,
398 : config->hcclCommName);
399 :
400 4 : CHK_RET(CallSingletons()); // 临时规避，在初始化通信域前声明单例保证时序
401 :
402 4 : HcclCommInfoV2 &opbasedCommInfoV2 = GetCommInfoV2();
403 4 : if (opbasedCommInfoV2.status == DeviceStatus::DEVICE_RECOVERED) {
404 1 : if (opbasedCommInfoV2.pComm->GetId() != config->hcclCommName) {
405 0 : HCCL_WARNING("[HcclCommInitClusterInfoConfig] Device was recovered, but communicator is search miss.");
406 : } else {
407 1 : *comm = static_cast<HcclComm>(opbasedCommInfoV2.pComm.get());
408 1 : return HCCL_SUCCESS;
409 : }
410 : }
411 :
412 9 : HCCL_INFO("HcclCommInitClusterInfoConfig ranktable[%s], config->hcclBufferSize[%u] MB",
413 : clusterInfo,
414 : config->hcclBufferSize);
415 3 : if (UNLIKELY(config->hcclBufferSize == 0)) {
416 0 : HCCL_ERROR("HcclCommInitClusterInfoConfigV2 config: hcclBufferSize is 0 MB, invalid para");
417 0 : return HCCL_E_PARA;
418 : }
419 : // 读取ranktable文件
420 3 : std::string ranktableM;
421 3 : CHK_RET(HcomLoadRankTableFileV2(clusterInfo, ranktableM));
422 :
423 3 : nlohmann::json data;
424 : try {
425 3 : data = nlohmann::json::parse(ranktableM);
426 0 : } catch (const nlohmann::json::parse_error &e) {
427 0 : HCCL_ERROR("[RankTable]JSON parse error: %s at byte %d", e.what(), e.byte);
428 0 : return HCCL_E_INTERNAL;
429 0 : } catch (const nlohmann::json::exception &e) {
430 0 : HCCL_ERROR("[RankTable]JSON parse error: %s", e.what());
431 0 : return HCCL_E_INTERNAL;
432 0 : } catch (...) {
433 0 : HCCL_ERROR("[RankTable]load allocated resource to json fail, please check json input");
434 0 : return HCCL_E_INTERNAL;
435 0 : };
436 :
437 3 : HcclResult ret = ParseJsonAndCreateComm(data, rank, config, comm, ranktableM);
438 :
439 3 : CHK_PRT_RET(ret, HCCL_ERROR("[Parse][Json]errNo[0x%016llx] and create comm failed.",
440 : HCCL_ERROR_CODE(ret)), static_cast<HcclResult>(ret));
441 :
442 : /* 关键状态记录 */
443 9 : HCCL_RUN_INFO("[HCCL_TRACE]%s success, take time [%lld]us, clusterInfo[%s], rank[%u], deviceLogicId[%d], devPhyId[%d], commId[%s].",
444 : __func__, DURATION_US(TIME_NOW() - startut), clusterInfo, rank, deviceLogicId, devPhyId, config->hcclCommName);
445 :
446 3 : return HCCL_SUCCESS;
447 3 : }
448 :
449 6 : HcclResult HcclCheckTaskServiceExist(const std::string &commId, s32 deviceId)
450 : {
451 6 : auto outerIt = g_taskServiceMap.find(commId);
452 6 : if (outerIt == g_taskServiceMap.end()) {
453 9 : HCCL_ERROR("[CheckTaskServiceExist] TaskService of CommId[%s] deviceId[%d],CommId Not Found", commId.c_str(), deviceId);
454 3 : return HCCL_E_NOT_FOUND;
455 : }
456 3 : auto innerIt = outerIt->second.find(deviceId);
457 3 : if (innerIt == outerIt->second.end()) {
458 9 : HCCL_ERROR("[CheckTaskServiceExist] TaskService of CommId[%s] deviceId[%d],deviceId Not Found", commId.c_str(), deviceId);
459 3 : return HCCL_E_NOT_FOUND;
460 : }
461 0 : return HCCL_SUCCESS;
462 : }
463 :
464 2 : HcclResult HcclTaskRegisterV2(HcclComm comm, const char *msgTag, Callback cb)
465 : {
466 6 : HCCL_RUN_INFO("[HcclTaskRegisterV2] start to register task");
467 2 : CHK_PTR_NULL(comm);
468 2 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
469 2 : std::string commId = communicator->GetId();
470 2 : s32 deviceId = communicator->GetDeviceLogicId();
471 8 : CHK_RET(HcclCheckTaskServiceExist(commId, deviceId));
472 :
473 0 : return g_taskServiceMap[commId][deviceId]->TaskRegister(msgTag, cb);
474 2 : }
475 :
476 2 : HcclResult HcclTaskRegisterProfV2(HcclComm comm, ProfCallbackTemplate profCallback)
477 : {
478 2 : CHK_PTR_NULL(comm);
479 2 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
480 2 : std::string commId = communicator->GetId();
481 6 : HCCL_INFO("[HcclTaskRegisterProfV2] commId[%s]", commId.c_str());
482 2 : s32 deviceId = communicator->GetDeviceLogicId();
483 8 : CHK_RET(HcclCheckTaskServiceExist(commId, deviceId));
484 0 : return g_taskServiceMap[commId][deviceId]->TaskProfRegister(profCallback);
485 2 : }
486 :
487 0 : HcclResult HcclGetDpuSteamIdV2(HcclComm comm, u32 &dpuStreamId) {
488 0 : HCCL_RUN_INFO("[HcclTaskRegisterV2] start to Get DpuSteamId");
489 0 : CHK_PTR_NULL(comm);
490 0 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
491 0 : auto ret = communicator->GetStreamId(dpuStreamId);
492 0 : if (ret != HCCL_SUCCESS) {
493 0 : HCCL_WARNING("[HcclGetDpuSteamIdV2] GetStreamId failed, ret[0x%016llx]", HCCL_ERROR_CODE(ret));
494 0 : return ret;
495 : }
496 0 : return HCCL_SUCCESS;
497 : }
498 :
499 2 : HcclResult HcclTaskUnRegisterV2(HcclComm comm, const char *msgTag)
500 : {
501 6 : HCCL_RUN_INFO("[HcclTaskUnRegisterV2] start to unregister task, g_taskServiceMap.size()==%zu", g_taskServiceMap.size());
502 2 : CHK_PTR_NULL(comm);
503 2 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
504 2 : std::string commId = communicator->GetId();
505 2 : s32 deviceId = communicator->GetDeviceLogicId();
506 8 : CHK_RET(HcclCheckTaskServiceExist(commId, deviceId));
507 0 : return g_taskServiceMap[commId][deviceId]->TaskUnRegister(msgTag);
508 2 : }
509 :
510 4 : HcclResult HcclGetRootInfoV2(HcclRootInfo *rootInfo)
511 : {
512 12 : HCCL_RUN_INFO("Entry-HcclGetRootInfo V950");
513 4 : HcclUs startut = TIME_NOW();
514 :
515 : // 执行root节点作为server端流程, 获得rootHandle
516 4 : HcclRootHandleV2 rootHandle{};
517 4 : std::shared_ptr<RankInfoDetect> rankInfoDetectServer;
518 4 : EXCEPTION_CATCH((rankInfoDetectServer = std::make_shared<RankInfoDetect>()), return HCCL_E_MEMORY);
519 10 : TRY_CATCH_RETURN(rankInfoDetectServer->SetupServer(rootHandle));
520 :
521 : // 先保活 server，避免后续校验失败时局部 shared_ptr 析构 join 卡住调用线程
522 2 : HcclCommInfoV2& opbasedCommInfoV2 = GetCommInfoV2();
523 2 : std::lock_guard<std::mutex> detectServerGuard(opbasedCommInfoV2.detectServerLock);
524 2 : opbasedCommInfoV2.hcclCommRankInfoDetectServer.insert({rootHandle.identifier, rankInfoDetectServer});
525 :
526 : // 校验rootHandle大小是否超过rootInfo->internal大小
527 2 : u32 rootHandleLen = sizeof(HcclRootHandleV2);
528 2 : CHK_PRT_RET(rootHandleLen > HCCL_ROOT_INFO_BYTES,
529 : HCCL_ERROR("[%s] hccl root info overflow. max length: %u, actual:%zu, identifier[%s]",
530 : __func__, HCCL_ROOT_INFO_BYTES, rootHandleLen, rootHandle.identifier), HCCL_E_INTERNAL);
531 :
532 : // 将rootHandle拷贝到rootInfo出参
533 2 : s32 sRet = memcpy_s(rootInfo->internal, HCCL_ROOT_INFO_BYTES, &rootHandle, rootHandleLen);
534 2 : CHK_PRT_RET(sRet != EOK, HCCL_ERROR("[%s] memcpy root info fail. errorno[%d] params:destMaxSize[%u],"
535 : " count[%u]", __func__, sRet, HCCL_ROOT_INFO_BYTES, rootHandleLen), HCCL_E_MEMORY);
536 :
537 : /* 首节点诊断信息记录 */
538 2 : s32 deviceLogicId = HcclGetThreadDeviceId();
539 2 : s32 devPhyId = HrtGetDevicePhyIdByIndex(deviceLogicId);
540 6 : HCCL_RUN_INFO("HcclGetRootInfoV2 success, take time [%lld]us, rootinfo: host ip[%s] port[%u] netMode[%s] "
541 : "identifier[%s], deviceLogicId[%d], devPhyId[%d]",
542 : DURATION_US(TIME_NOW() - startut), rootHandle.ip, rootHandle.listenPort,
543 : rootHandle.netMode.Describe().c_str(), rootHandle.identifier, deviceLogicId, devPhyId);
544 2 : return HCCL_SUCCESS;
545 4 : }
546 :
547 1 : HcclResult GetDeviceCommV2(uint32_t ndev, const HcclRootInfo &rootHandle, const s32 rank, const s32 logicDeviceId,
548 : HcclComm &comm)
549 : {
550 : //给当前线程添加名字
551 1 : SetThreadName("Hccl_GetDevComm");
552 :
553 1 : TRY_CATCH_RETURN(HrtSetDevice(logicDeviceId));
554 1 : std::string identifier;
555 1 : HcclResult ret = HcclCommInitRootInfoV2(ndev, &rootHandle, rank, &comm, identifier);
556 1 : if (ret != HCCL_SUCCESS || comm == nullptr) {
557 0 : comm = nullptr;
558 0 : HCCL_ERROR("[GetDeviceComm] rank[%d] Get device comm failed, ret[%d]!", rank, ret);
559 0 : TRY_CATCH_RETURN(HrtSetDevice(logicDeviceId));
560 0 : return ret;
561 : }
562 :
563 1 : return HCCL_SUCCESS;
564 1 : }
565 :
566 2 : HcclResult HcclGetCommAllV2(uint32_t ndev, int32_t *devices, HcclComm *comms)
567 : {
568 : // 入参校验
569 2 : CHK_PRT_RET(ndev == 0, HCCL_ERROR("[HcclGetCommAll] ndev is invalid ndev[%u]", ndev), HCCL_E_PARA);
570 2 : CHK_PTR_NULL(comms);
571 2 : CHK_PTR_NULL(devices);
572 :
573 : //给当前线程添加名字
574 2 : SetThreadName("Hccl_GetCommAllV2");
575 :
576 2 : TRY_CATCH_RETURN(HrtSetDevice(devices[0]));
577 :
578 : // 获取通信域之前, 先把所有通信域设置为空
579 4 : for (uint32_t i = 0; i < ndev; i++) {
580 2 : comms[i] = nullptr;
581 : }
582 :
583 : HcclRootInfo rootHandle;
584 5 : CHK_RET(HcclGetRootInfoV2(&rootHandle));
585 :
586 1 : std::vector<std::unique_ptr<std::thread>> threads(ndev);
587 2 : for (uint32_t rankId = 0; rankId < ndev; rankId++) {
588 2 : threads[rankId].reset(new (std::nothrow) std::thread(&GetDeviceCommV2, ndev, std::ref(rootHandle), rankId,
589 1 : devices[rankId], std::ref(comms[rankId])));
590 1 : CHK_PRT_RET(!threads[rankId], HCCL_ERROR("[HcclGetCommAllV2]threads[%u] reset failed ", rankId), HCCL_E_INTERNAL);
591 : }
592 2 : for (uint32_t i = 0; i < ndev; i++) {
593 1 : threads[i]->join();
594 : }
595 :
596 : // 如果任何一个通信域初始化失败，将所有已经成功创建的通信域销毁
597 1 : bool isFailed = false;
598 2 : for (uint32_t i = 0; i < ndev; ++i) {
599 1 : if (comms[i] == nullptr) {
600 0 : HCCL_ERROR("[HcclGetCommAllV2] rank[%u] get comm failed!", i);
601 0 : isFailed = true;
602 0 : break;
603 : }
604 : }
605 1 : if (isFailed) {
606 0 : for (uint32_t i = 0; i < ndev; ++i) {
607 0 : if (comms[i] != nullptr) {
608 0 : (void)HcclCommDestroyV2(comms[i]);
609 : }
610 : }
611 0 : return HCCL_E_INTERNAL;
612 : }
613 :
614 1 : TRY_CATCH_RETURN(HrtSetDevice(devices[0]));
615 :
616 1 : return HCCL_SUCCESS;
617 1 : }
618 :
619 2 : HcclResult HcclCommInitAllV2(uint32_t ndev, int32_t *devices, HcclComm *comms)
620 : {
621 2 : HcclUs startut = TIME_NOW();
622 2 : std::string devicesStr;
623 4 : for (size_t i = 0; i < ndev; ++i) {
624 2 : std::string deviceStr = std::to_string(devices[i]);
625 2 : devicesStr += deviceStr;
626 2 : if (i != ndev - 1) {
627 0 : devicesStr += " ";
628 : }
629 2 : }
630 6 : HCCL_RUN_INFO("Entry-HcclCommInitAll V950, ndev:[%u], devices:[%s].", ndev, devicesStr.c_str());
631 :
632 2 : std::future<HcclResult> threadResult;
633 2 : std::unique_ptr<std::thread> getCommThread;
634 2 : getCommThread.reset(new (std::nothrow) std::thread(
635 4 : [=, &threadResult]() { threadResult = std::async(std::launch::async, HcclGetCommAllV2, ndev, devices, comms); }));
636 2 : CHK_PRT_RET(!getCommThread, HCCL_ERROR("[HcclCommInitAll]thread reset failed "), HCCL_E_INTERNAL);
637 2 : getCommThread->join();
638 :
639 2 : HcclResult ret = threadResult.get();
640 2 : if (ret != HCCL_SUCCESS) {
641 2 : for (uint32_t i = 0; i < ndev; ++i) {
642 1 : if (comms[i] != nullptr) {
643 0 : (void)HcclCommDestroyV2(comms[i]);
644 0 : comms[i] = nullptr;
645 : }
646 : }
647 3 : HCCL_ERROR("HcclCommInitAll failed! threadResult[%d]", ret);
648 1 : return ret;
649 : }
650 1 : s32 deviceLogicId = HcclGetThreadDeviceId();
651 1 : s32 devPhyId = HrtGetDevicePhyIdByIndex(deviceLogicId);
652 3 : HCCL_RUN_INFO("HcclCommInitAll success, take time [%lld]us, deviceLogicId[%d], devPhyId[%d].", DURATION_US(TIME_NOW() - startut),
653 : deviceLogicId, devPhyId);
654 1 : return HCCL_SUCCESS;
655 2 : }
656 :
657 5 : HcclResult HcclCommDestroyV2(HcclComm comm)
658 : {
659 5 : HcclUs startut = TIME_NOW();
660 5 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
661 5 : CHK_PTR_NULL(communicator);
662 5 : string commId = communicator->GetId();
663 5 : HcclCommInfoV2 &opbasedCommInfoV2 = GetCommInfoV2();
664 15 : HCCL_RUN_INFO("Entry-HcclCommDestroy V950 comm[%s]", commId.c_str());
665 :
666 5 : if (communicator->GetCommStatus() == CommStatus::COMM_INUSE) {
667 3 : HCCL_WARNING("[HcclCommDestroy] comm is in use, please try again later");
668 1 : return HCCL_E_AGAIN;
669 : }
670 4 : std::unique_lock<std::mutex> lock(opbasedCommInfoV2.groupParamsLock);
671 4 : if (commId == opbasedCommInfoV2.commParams.commId && opbasedCommInfoV2.pComm != nullptr) {
672 : // 通信域销毁，更新子通信域ccu使用情况
673 1 : opbasedCommInfoV2.ccuStatus.RemoveCommId(opbasedCommInfoV2.pComm->GetId());
674 2 : for (auto iterGroup : opbasedCommInfoV2.hcclGroupMap) {
675 1 : opbasedCommInfoV2.ccuStatus.RemoveCommId(iterGroup.first);
676 1 : }
677 1 : opbasedCommInfoV2.pComm = nullptr;
678 1 : opbasedCommInfoV2.status = DeviceStatus::DEVICE_IDLE;
679 : }
680 : // 通信域销毁，更新ccu使用情况
681 4 : opbasedCommInfoV2.ccuStatus.RemoveCommId(commId);
682 4 : auto iter = opbasedCommInfoV2.hcclGroupMap.find(commId);
683 4 : if (iter != opbasedCommInfoV2.hcclGroupMap.end()) {
684 1 : opbasedCommInfoV2.hcclGroupMap.erase(commId); // 删除通信域实例，在communicatorImpl类的析构函数中有dpu与npu之间共享内存的销毁
685 : // 通信域销毁，更新ccu使用情况
686 1 : opbasedCommInfoV2.ccuStatus.RemoveCommId(commId);
687 : } else {
688 3 : s32 deviceLogicId = HcclGetThreadDeviceId();
689 9 : HCCL_ERROR("[HcclCommDestroyV2] comm is not exist, comm=%p, group=%s, deviceLogicId=%d",
690 : comm, commId.c_str(), deviceLogicId);
691 3 : return HCCL_E_PARA;
692 : }
693 1 : lock.unlock();
694 :
695 1 : s32 deviceLogicId = HcclGetThreadDeviceId();
696 1 : s32 devPhyId = HrtGetDevicePhyIdByIndex(deviceLogicId);
697 3 : HCCL_RUN_INFO("HcclCommDestroy V950 comm[%s] success, take time [%lld]us, deviceLogicId[%d], devPhyId[%d].", commId.c_str(),
698 : DURATION_US(TIME_NOW() - startut), deviceLogicId, devPhyId);
699 1 : return HCCL_SUCCESS;
700 5 : }
701 :
702 2 : HcclResult HcclAlltoAllV2(const void *sendBuf, uint64_t sendCount, HcclDataType sendType, const void *recvBuf,
703 : uint64_t recvCount, HcclDataType recvType, HcclComm comm, aclrtStream stream)
704 : {
705 2 : HcclUs startut = TIME_NOW();
706 : bool isCapture;
707 2 : rtModel_t rtModel = nullptr;
708 2 : u32 modelId = 0;
709 :
710 2 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
711 2 : const std::string tag = "ALLTOALL_" + communicator->GetId();
712 :
713 2 : CHK_RET(HcomCheckOpParamV2(tag.c_str(), 0, sendType, stream));
714 2 : CHK_RET(HcomCheckDataTypeV2(recvType));
715 2 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
716 :
717 : // 接口交互信息日志
718 : char stackLogBufferV2[LOG_TMPBUF_SIZE];
719 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
720 1 : s32 streamId = HrtGetStreamId(stream);
721 1 : s32 deviceLogicId = HrtGetDevice();
722 1 : u32 localRank = INVALID_VALUE_RANKID;
723 1 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
724 :
725 2 : s32 ret = snprintf_s(stackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
726 : "tag[%s], sendBuf[%p], recvBuf[%p], sendCount[%llu], recvCount[%llu], sendType[%s],"
727 : "recvType[%s], localRank[%u], streamId[%d], deviceLogicId[%d]",
728 2 : tag.c_str(), sendBuf, recvBuf, sendCount, recvCount, GetDataTypeEnumStrV2(sendType).c_str(),
729 2 : GetDataTypeEnumStrV2(recvType).c_str(), localRank, streamId, deviceLogicId);
730 :
731 1 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
732 1 : std::string logInfo = "Entry-HcclAlltoAllV2:" + std::string(stackLogBufferV2);
733 1 : if (isCapture) {
734 0 : CHK_PTR_NULL(rtModel);
735 : // 获取不到modelId会报错
736 0 : CHK_RET(GetModelId(rtModel, modelId));
737 0 : logInfo += ", model id[" + to_string(modelId) + "].";
738 : }
739 1 : communicator->GetTrace().Save(logInfo);
740 1 : }
741 :
742 2 : static thread_local Hccl::CollOpParams opParams;
743 2 : opParams.opType = Hccl::OpType::ALLTOALL;
744 2 : opParams.sendBuf = const_cast<void *>(sendBuf);
745 2 : opParams.recvBuf = const_cast<void *>(recvBuf);
746 2 : opParams.all2AllDataDes.sendCount = sendCount;
747 2 : opParams.all2AllDataDes.recvCount = recvCount;
748 2 : opParams.all2AllDataDes.sendType = HcclDataTypeToDataType(sendType);
749 2 : opParams.all2AllDataDes.recvType = HcclDataTypeToDataType(recvType);
750 2 : opParams.dataType = HcclDataTypeToDataType(sendType);
751 2 : opParams.opTag = tag;
752 :
753 2 : CHK_RET_AND_PRINT_IDE(communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream)), tag.c_str());
754 :
755 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
756 1 : HcclUs endut = TIME_NOW();
757 : /* 关键状态记录 */
758 1 : std::string endInfo = "HcclAlltoAllV2:success,take time: " +
759 2 : std::to_string(DURATION_US(endut - startut).count()) + " us," + std::string(stackLogBufferV2);
760 1 : communicator->GetTrace().Save(endInfo);
761 1 : }
762 :
763 2 : return HCCL_SUCCESS;
764 2 : }
765 :
766 2 : HcclResult HcclAlltoAllVV2(const void *sendBuf, const void *sendCounts, const void *sdispls, HcclDataType sendType,
767 : const void *recvBuf, const void *recvCounts, const void *rdispls, HcclDataType recvType, HcclComm comm,
768 : aclrtStream stream)
769 : {
770 2 : HcclUs startut = TIME_NOW();
771 : bool isCapture;
772 2 : rtModel_t rtModel = nullptr;
773 2 : u32 modelId = 0;
774 :
775 2 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
776 2 : const std::string tag = "HCCL_ALLTOALLV_" + communicator->GetId();
777 :
778 2 : CHK_RET_AND_PRINT_IDE(HcomCheckOpParamV2(tag.c_str(), 0, sendType, stream), tag.c_str());
779 2 : CHK_RET_AND_PRINT_IDE(HcomCheckDataTypeV2(recvType), tag.c_str());
780 2 : CHK_RET(HcomCheckDataTypeV2(sendType));
781 2 : CHK_RET(HcomCheckDataTypeV2(recvType));
782 2 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
783 : /* 根据ranksize校验相关入参 */
784 2 : u32 rankSize = 0;
785 2 : CHK_RET(communicator->GetRankSize(&rankSize));
786 2 : CHK_RET(HcomCheckAlltoAllVExternalMemV2(sendBuf, sendCounts, recvBuf, recvCounts, rankSize));
787 :
788 : // 接口交互信息日志
789 : char stackLogBufferV2[LOG_TMPBUF_SIZE];
790 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
791 1 : s32 streamId = HrtGetStreamId(stream);
792 1 : s32 deviceLogicId = HrtGetDevice();
793 1 : u32 localRank = INVALID_VALUE_RANKID;
794 1 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
795 :
796 2 : s32 ret = snprintf_s(stackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
797 : "tag[%s], sendBuf[%p], recvBuf[%p], sendCounts[%p], recvCounts[%p], sendType[%s],"
798 : "recvType[%s], localRank[%u], streamId[%d], deviceLogicId[%d]",
799 2 : tag.c_str(), sendBuf, recvBuf, sendCounts, recvCounts, GetDataTypeEnumStrV2(sendType).c_str(),
800 2 : GetDataTypeEnumStrV2(recvType).c_str(), localRank, streamId, deviceLogicId);
801 :
802 1 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
803 1 : std::string logInfo = "Entry-HcclAlltoAllVV2:" + std::string(stackLogBufferV2);
804 1 : if (isCapture) {
805 0 : CHK_PTR_NULL(rtModel);
806 : // 获取不到modelId会报错
807 0 : CHK_RET(GetModelId(rtModel, modelId));
808 0 : logInfo += ", model id[" + to_string(modelId) + "].";
809 : }
810 1 : communicator->GetTrace().Save(logInfo);
811 1 : }
812 :
813 2 : static thread_local Hccl::CollOpParams opParams;
814 2 : opParams.opType = Hccl::OpType::ALLTOALLV;
815 2 : opParams.sendBuf = const_cast<void *>(sendBuf);
816 2 : opParams.recvBuf = const_cast<void *>(recvBuf);
817 2 : opParams.all2AllVDataDes.sendCounts = const_cast<void *>(sendCounts);
818 2 : opParams.all2AllVDataDes.recvCounts = const_cast<void *>(recvCounts);
819 2 : opParams.all2AllVDataDes.sdispls = const_cast<void *>(sdispls);
820 2 : opParams.all2AllVDataDes.rdispls = const_cast<void *>(rdispls);
821 2 : opParams.all2AllVDataDes.sendType = HcclDataTypeToDataType(sendType);
822 2 : opParams.all2AllVDataDes.recvType = HcclDataTypeToDataType(recvType);
823 2 : opParams.dataType = HcclDataTypeToDataType(sendType);
824 2 : opParams.opTag = tag;
825 :
826 2 : CHK_RET_AND_PRINT_IDE(communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream)), tag.c_str());
827 :
828 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
829 1 : HcclUs endut = TIME_NOW();
830 : /* 关键状态记录 */
831 1 : std::string endInfo = "HcclAlltoAllVV2:success,take time: " +
832 2 : std::to_string(DURATION_US(endut - startut).count()) + " us," + std::string(stackLogBufferV2);
833 1 : communicator->GetTrace().Save(endInfo);
834 1 : }
835 :
836 2 : return HCCL_SUCCESS;
837 2 : }
838 :
839 : // 单算子 创建子通信域 V2
840 3 : HcclResult HcclCreateSubCommConfigV2(const HcclComm *comm, uint32_t rankNum, uint32_t *rankIds, uint64_t subCommId,
841 : uint32_t subCommRankId, HcclCommConfig *config, HcclComm *subComm)
842 : {
843 3 : HcclUs startut = TIME_NOW();
844 :
845 3 : std::ostringstream printRankIds;
846 3 : unordered_set<uint32_t> rankIdSet;
847 3 : printRankIds << "input rankIds: ";
848 6 : for (u32 i = 0; i < rankNum; i++) {
849 3 : CHK_PTR_NULL(rankIds + i);
850 3 : printRankIds << "rank[";
851 3 : printRankIds << i;
852 3 : printRankIds << "] = ";
853 3 : printRankIds << rankIds[i];
854 3 : if (i < rankNum - 1) {
855 0 : printRankIds << ", ";
856 : }
857 3 : CHK_PRT_RET(
858 : rankIdSet.find(rankIds[i]) != rankIdSet.end(),
859 : HCCL_ERROR("[GetHcomRankListV2]errNo[0x%016llx], " \
860 : "duplicated rankId[%u] in rankIds.",
861 : HCCL_ERROR_CODE(HCCL_E_PARA), rankIds[i]),
862 : HCCL_E_PARA);
863 3 : rankIdSet.insert(rankIds[i]);
864 : }
865 :
866 9 : HCCL_RUN_INFO("Entry-HcclCreateSubCommConfig V950 rankIds[%s], subCommRankId[%u], commEngine[%u], hcclBufferSize[%u] MB",
867 : printRankIds.str().c_str(), subCommRankId, config->hcclOpExpansionMode, config->hcclBufferSize);
868 :
869 3 : HcclCommInfoV2 &opbasedCommInfoV2 = GetCommInfoV2();
870 3 : if (opbasedCommInfoV2.status == DeviceStatus::DEVICE_RECOVERED) {
871 3 : if (opbasedCommInfoV2.hcclGroupMap.find(config->hcclCommName) == opbasedCommInfoV2.hcclGroupMap.end()) {
872 0 : HCCL_WARNING("[HcclCommInitClusterInfoConfig] Device was recovered, but communicator is search miss.");
873 : } else {
874 2 : *subComm = static_cast<HcclComm>(opbasedCommInfoV2.hcclGroupMap[config->hcclCommName].pComm.get());
875 1 : return HCCL_SUCCESS;
876 : }
877 : }
878 6 : HCCL_RUN_INFO("Entry-HcclCreateSubCommConfig V950 config->hcclBufferSize[%u] MB", config->hcclBufferSize);
879 :
880 2 : CHK_PRT_RET(UNLIKELY(config->hcclBufferSize == 0),
881 : HCCL_ERROR("HcclCreateSubCommConfigV2 config: hcclBufferSize is 0 MB, invalid para"),
882 : HCCL_E_PARA);
883 2 : if (config->hcclBufferSize == HCCL_COMM_BUFFSIZE_CONFIG_NOT_SET) {
884 1 : config->hcclBufferSize = 0;
885 3 : HCCL_INFO("[HcclCreateSubCommConfigV2] set default HCCL BUFFER is 200 MB");
886 : }
887 2 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(*comm);
888 :
889 2 : string commId = communicator->GetId();
890 2 : if (!communicator->IsWorldGroup()) {
891 0 : HCCL_ERROR("[HcclCreateSubCommConfig] commId [%s] is not HCCL WORLD GROUP", commId.c_str());
892 0 : return HCCL_E_INTERNAL;
893 : }
894 :
895 2 : CHK_PRT_RET(opbasedCommInfoV2.pComm == nullptr,
896 : HCCL_ERROR("[Create][Group]HcclCommInfoV2.pComm is null, please check if the initialize process is called."),
897 : HCCL_E_PTR);
898 :
899 2 : std::unique_lock<std::mutex> groupParaLock(opbasedCommInfoV2.groupParamsLock);
900 :
901 2 : string subCommIdStr(commId + "_sub_" + to_string(subCommId));
902 2 : LoadConfigCommName(subCommIdStr, *config);
903 2 : if (subCommIdStr == commId) {
904 0 : HCCL_ERROR("[HcclCreateSubCommConfigV2] sub comm name[%s] should not be same as the world comm name[%s]",
905 : subCommIdStr.c_str(), commId.c_str());
906 0 : return HCCL_E_INTERNAL;
907 : }
908 :
909 : /* 已经存在的group不允许再次创建 */
910 2 : if (opbasedCommInfoV2.hcclGroupMap.find(subCommIdStr) != opbasedCommInfoV2.hcclGroupMap.end()) {
911 0 : HCCL_ERROR("[Create][Group]errNo[0x%016llx] group[%s] is already exist",
912 : HCCL_ERROR_CODE(HCCL_E_PARA), subCommIdStr.c_str());
913 0 : return HCCL_E_PARA;
914 : }
915 :
916 : /* 创建groupParamsV2Tem */
917 2 : HcclGroupParamsV2 groupParamsV2Tem;
918 2 : CHK_RET(GetHcomRankListV2(rankNum, rankIds, groupParamsV2Tem));
919 :
920 : /* 如果是groupRank = INVALID_VALUE_RANKID，即本rank不参与create group */
921 2 : if (groupParamsV2Tem.groupRank == INVALID_VALUE_RANKID) {
922 0 : HCCL_ERROR("[Create][Group]errNo[0x%016llx] confirm groupRank from worldRank[%d] error",
923 : HCCL_ERROR_CODE(HCCL_E_NOT_FOUND),
924 : opbasedCommInfoV2.commParams.myRank);
925 0 : return HCCL_E_NOT_FOUND;
926 : }
927 :
928 : /* 创建子通信域 */
929 : Hccl::CommParams commParams{subCommIdStr, static_cast<Hccl::RankId>(subCommRankId),
930 2 : rankNum, opbasedCommInfoV2.commParams.myRank, opbasedCommInfoV2.commParams.devType};
931 2 : CheckHcclDeterministic(config->hcclDeterministic);
932 : // 默认全都开启确定性计算
933 2 : config->hcclDeterministic = 1;
934 2 : HcclCommConfig hcclConf = *config;
935 : std::shared_ptr<Hccl::HcclCommunicator> subCommunicator =
936 2 : make_shared<Hccl::HcclCommunicator>(commParams, &hcclConf);
937 :
938 2 : std::vector<u32> rankIdsVec(rankNum);
939 4 : for (uint32_t i = 0; i < rankNum; ++i) {
940 2 : rankIdsVec[i] = rankIds[i];
941 : }
942 : /* --------------初始化------------------------- */
943 2 : HcclResult ret = HCCL_SUCCESS;
944 2 : bool errorFlag = false;
945 2 : s32 logicDevId = HrtGetDevice();
946 2 : s32 devPhyId = HrtGetDevicePhyIdByIndex(logicDevId);
947 : do {
948 2 : ret = communicator->CreateSubComm(commParams, rankIdsVec, subCommunicator, hcclConf);
949 2 : CHK_PRT_BREAK(ret != HcclResult::HCCL_SUCCESS,
950 : HCCL_ERROR("[%s]communicator->CreateSubComm failed, errNo[0x%016llx]", __func__, HCCL_ERROR_CODE(ret)),
951 : errorFlag = true);
952 2 : CHK_SMART_PTR_NULL(subCommunicator);
953 :
954 2 : subCommunicator->RegisterAcceStateCallBack(CommunicatorCallback());
955 :
956 2 : groupParamsV2Tem.pComm = subCommunicator;
957 :
958 2 : opbasedCommInfoV2.hcclGroupMap.insert(std::make_pair(subCommIdStr, groupParamsV2Tem));
959 2 : groupParaLock.unlock();
960 :
961 2 : ret = CommManager::GetInstance(logicDevId).SetCommAcceleratorV2(subCommunicator.get(), config->hcclOpExpansionMode); // 通信域创建，设置默认accelerator
962 2 : CHK_PRT_BREAK(ret != HcclResult::HCCL_SUCCESS,
963 : HCCL_ERROR("[%s]SetCommAcceleratorV2 failed, errNo[0x%016llx]", __func__, HCCL_ERROR_CODE(ret)),
964 : errorFlag = true);
965 2 : *subComm = subCommunicator.get();
966 : } while (0);
967 :
968 2 : if (errorFlag) {
969 0 : HCCL_ERROR("[Init][%s]HcclCreateSubCommConfigV2 failed, deviceLogicId[%d], devPhyId[%d], sub comm[%s], world comm[%s]"\
970 : "return[0x%016llx]", __func__,
971 : logicDevId, devPhyId, subCommIdStr.c_str(), commId.c_str(), HCCL_ERROR_CODE(ret));
972 0 : (void)HcclCommDestroyV2(subCommunicator.get());
973 0 : *subComm = nullptr;
974 0 : return ret;
975 : }
976 : /* 关键状态记录 */
977 6 : HCCL_RUN_INFO("[Create][Group]create group[%s] success, deviceLogicId[%d], devPhyId[%d], take time [%lld]us",
978 : subCommIdStr.c_str(), logicDevId, devPhyId, DURATION_US(TIME_NOW() - startut));
979 2 : return HCCL_SUCCESS;
980 3 : }
981 :
982 1 : HcclResult HcclGetRankIdV2(HcclComm comm, uint32_t *rank)
983 : {
984 3 : HCCL_RUN_INFO("Entry-HcclGetRankId V950");
985 1 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
986 1 : auto ret = communicator->GetRankId(*rank);
987 1 : if (ret != HcclResult::HCCL_SUCCESS) {
988 0 : return HCCL_E_INTERNAL;
989 : }
990 : /* 关键状态记录 */
991 3 : HCCL_RUN_INFO("Entry-HcclGetRankId V950 success, comm[%s], rankIdPtr[%p], rankId[%u]",
992 : communicator->GetId().c_str(), rank, *rank);
993 1 : return HCCL_SUCCESS;
994 : }
995 :
996 1 : HcclResult HcclGetCommNameV2(HcclComm commHandle, char *commName)
997 : {
998 1 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(commHandle);
999 1 : if (communicator == nullptr) {
1000 0 : HCCL_ERROR("HcclGetCommNameV2 communicator is nullptr");
1001 0 : return HCCL_E_PTR;
1002 : }
1003 3 : HCCL_INFO("HcclGetCommNameV2 commId[%s], commId size[%zu], input commName ptr=[%p]",
1004 : communicator->GetId().c_str(), communicator->GetId().size(), commName);
1005 1 : s32 ret = strncpy_s(
1006 1 : commName, ROOTINFO_INDENTIFIER_MAX_LENGTH, communicator->GetId().c_str(), communicator->GetId().size() + 1);
1007 1 : CHK_PRT_RET(ret != EOK, HCCL_ERROR("HcclGetCommName str copy fail. return[%d], commId[%s]", ret, communicator->GetId().c_str()),
1008 : HCCL_E_INTERNAL);
1009 1 : return HCCL_SUCCESS;
1010 : }
1011 :
1012 9 : HcclResult HcclGetRankSizeV2(HcclComm comm, uint32_t *rankSize)
1013 : {
1014 9 : CHK_PTR_NULL(comm);
1015 9 : CHK_PTR_NULL(rankSize);
1016 9 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
1017 9 : s32 deviceLogicId = HcclGetThreadDeviceId();
1018 9 : s32 devPhyId = HrtGetDevicePhyIdByIndex(deviceLogicId);
1019 27 : HCCL_RUN_INFO("Entry-HcclGetRankSize V950, commId[%s], deviceLogicId[%d], devPhyId[%d]", communicator->GetId().c_str(),
1020 : deviceLogicId, devPhyId);
1021 9 : auto ret = communicator->GetRankSize(rankSize);
1022 9 : if (ret != HCCL_SUCCESS) {
1023 0 : HCCL_ERROR("HcclGetRankSizeV2 failed, rankSize[%u], commId[%s]", *rankSize, communicator->GetId().c_str());
1024 0 : return HCCL_E_INTERNAL;
1025 : }
1026 : /* 关键状态记录 */
1027 27 : HCCL_RUN_INFO("Entry-HcclGetRankSize V950 success, comm[%s], rankSizePtr[%p], rankSize[%u]",
1028 : communicator->GetId().c_str(), rankSize, *rankSize);
1029 9 : return HCCL_SUCCESS;
1030 : }
1031 :
1032 2 : HcclResult HcclAlltoAllVCV2(const void *sendBuf, const void *sendCountMatrix, HcclDataType sendType,
1033 : const void *recvBuf, HcclDataType recvType, HcclComm comm, rtStream_t stream)
1034 : {
1035 2 : HcclUs startut = TIME_NOW();
1036 : bool isCapture;
1037 2 : rtModel_t rtModel = nullptr;
1038 2 : u32 modelId = 0;
1039 2 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
1040 :
1041 2 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
1042 2 : const std::string tag = "HCCL_ALLTOALLVC_" + communicator->GetId();
1043 :
1044 2 : CHK_RET_AND_PRINT_IDE(HcomCheckOpParamV2(tag.c_str(), 0, sendType, stream), tag.c_str());
1045 2 : CHK_RET_AND_PRINT_IDE(HcomCheckDataTypeV2(recvType), tag.c_str());
1046 2 : u32 rankSize = 0;
1047 2 : CHK_RET(communicator->GetRankSize(&rankSize));
1048 2 : u32 myRank = INVALID_VALUE_RANKID;
1049 2 : CHK_RET(communicator->GetRankId(myRank));
1050 2 : bool isEmpty = false;
1051 2 : CHK_RET(HcomCheckAlltoAllVCEmptyV2(sendBuf, sendCountMatrix, recvBuf, rankSize, isEmpty));
1052 2 : if(isEmpty) {
1053 6 : HCCL_INFO("[HcclAlltoAllVCV2] sendCountMatrix is Empty");
1054 2 : return HCCL_SUCCESS;
1055 : }
1056 0 : CHK_RET(HcomCheckAlltoAllVCExternalMemV2(sendBuf, sendCountMatrix, recvBuf, rankSize, myRank));
1057 :
1058 : u64 sendCountMatrixHash;
1059 0 : HcomGetHashFromSendCountMatrixV2(sendCountMatrixHash, sendCountMatrix, rankSize, tag);
1060 : /* 接口交互信息日志 */
1061 : char stackLogBufferV2[LOG_TMPBUF_SIZE];
1062 0 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1063 0 : s32 streamId = HrtGetStreamId(stream);
1064 0 : s32 deviceLogicId = HrtGetDevice();
1065 0 : u32 localRank = INVALID_VALUE_RANKID;
1066 0 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
1067 :
1068 0 : s32 ret = snprintf_s(stackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
1069 : "tag[%s], sendBuf[%p], sendCountMatrixHash[%llu], sendType[%s], recvBuf[%p],"
1070 : "recvType[%s], localRank[%u], streamId[%d], deviceLogicId[%d]",
1071 0 : tag.c_str(), sendBuf, sendCountMatrixHash, GetDataTypeEnumStrV2(sendType).c_str(), recvBuf,
1072 0 : GetDataTypeEnumStrV2(recvType).c_str(), localRank, streamId, deviceLogicId);
1073 :
1074 0 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
1075 0 : std::string logInfo = "Entry-HcclAlltoAllVCV2:" + std::string(stackLogBufferV2);
1076 0 : if (isCapture) {
1077 0 : CHK_PTR_NULL(rtModel);
1078 : // 获取不到modelId会报错
1079 0 : CHK_RET(GetModelId(rtModel, modelId));
1080 0 : logInfo += ", model id[" + to_string(modelId) + "].";
1081 : }
1082 0 : communicator->GetTrace().Save(logInfo);
1083 0 : }
1084 :
1085 0 : static thread_local Hccl::CollOpParams opParams;
1086 0 : opParams.opType = Hccl::OpType::ALLTOALLVC;
1087 0 : opParams.sendBuf = const_cast<void *>(sendBuf);
1088 0 : opParams.recvBuf = const_cast<void *>(recvBuf);
1089 0 : opParams.all2AllVCDataDes.sendCountMatrix = const_cast<void *>(sendCountMatrix);
1090 0 : opParams.all2AllVCDataDes.sendType = HcclDataTypeToDataType(sendType);
1091 0 : opParams.all2AllVCDataDes.recvType = HcclDataTypeToDataType(recvType);
1092 0 : opParams.dataType = HcclDataTypeToDataType(sendType);
1093 0 : CHK_RET_AND_PRINT_IDE(communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream)), tag.c_str());
1094 :
1095 0 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1096 0 : HcclUs endut = TIME_NOW();
1097 : /* 关键状态记录 */
1098 0 : std::string endInfo = "HcclAlltoAllVCV2:success,take time: " +
1099 0 : std::to_string(DURATION_US(endut - startut).count()) + " us," + std::string(stackLogBufferV2);
1100 0 : communicator->GetTrace().Save(endInfo);
1101 0 : }
1102 :
1103 0 : return HCCL_SUCCESS;
1104 2 : }
1105 :
1106 5 : HcclResult HcclReduceV2(void *sendBuf, void *recvBuf, uint64_t count, HcclDataType dataType, HcclReduceOp op,
1107 : uint32_t root, HcclComm comm, aclrtStream stream)
1108 : {
1109 5 : HcclUs startut = TIME_NOW();
1110 : bool isCapture;
1111 5 : rtModel_t rtModel = nullptr;
1112 5 : u32 modelId = 0;
1113 :
1114 5 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
1115 5 : const std::string tag = "Reduce_" + communicator->GetId();
1116 :
1117 5 : CHK_RET_AND_PRINT_IDE(HcomCheckOpParamV2(tag.c_str(), count, dataType, stream), tag.c_str());
1118 5 : CHK_RET_AND_PRINT_IDE(HcomCheckReductionOpV2(op), tag.c_str());
1119 11 : CHK_RET_AND_PRINT_IDE(HcomCheckReduceDataTypeV2(dataType, op), tag.c_str());
1120 4 : u32 rankSize = INVALID_VALUE_RANKSIZE;
1121 4 : CHK_RET_AND_PRINT_IDE(communicator->GetRankSize(&rankSize), tag.c_str());
1122 4 : CHK_RET_AND_PRINT_IDE(HcomCheckUserRankV2(rankSize, root), tag.c_str());
1123 4 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
1124 :
1125 : /* 接口交互信息日志 */
1126 : char stackLogBufferV2[LOG_TMPBUF_SIZE];
1127 4 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1128 1 : s32 streamId = HrtGetStreamId(stream);
1129 1 : s32 deviceLogicId = HrtGetDevice();
1130 1 : u32 localRank = INVALID_VALUE_RANKID;
1131 1 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
1132 :
1133 2 : s32 ret = snprintf_s(stackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
1134 : "tag[%s], sendBuf[%p], recvBuf[%p], count[%llu], dataType[%s], op[%s], root[%u],"
1135 : "localRank[%u], streamId[%d], deviceLogicId[%d]",
1136 3 : tag.c_str(), sendBuf, recvBuf, count, GetDataTypeEnumStrV2(dataType).c_str(), GetReduceOpEnumStrV2(op).c_str(),
1137 : root, localRank, streamId, deviceLogicId);
1138 :
1139 1 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
1140 1 : std::string logInfo = "Entry-HcclReduceV2:" + std::string(stackLogBufferV2);
1141 1 : if (isCapture) {
1142 0 : CHK_PTR_NULL(rtModel);
1143 : // 获取不到modelId会报错
1144 0 : CHK_RET(GetModelId(rtModel, modelId));
1145 0 : logInfo += ", model id[" + to_string(modelId) + "].";
1146 : }
1147 1 : communicator->GetTrace().Save(logInfo);
1148 1 : }
1149 :
1150 4 : static thread_local Hccl::CollOpParams opParams;
1151 4 : opParams.opType = Hccl::OpType::REDUCE;
1152 4 : opParams.dataType = HcclDataTypeToDataType(dataType);
1153 4 : opParams.reduceOp = HCCL_OP_REDUCE_MAP[op];
1154 4 : opParams.sendBuf = sendBuf;
1155 4 : opParams.recvBuf = recvBuf;
1156 4 : opParams.count = count;
1157 4 : opParams.root = root;
1158 4 : opParams.opTag = tag;
1159 4 : CHK_RET_AND_PRINT_IDE(communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream)), tag.c_str());
1160 :
1161 4 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1162 1 : HcclUs endut = TIME_NOW();
1163 : /* 关键状态记录 */
1164 1 : std::string endInfo = "HcclReduceV2:success,take time: " +
1165 2 : std::to_string(DURATION_US(endut - startut).count()) + " us," + std::string(stackLogBufferV2);
1166 1 : communicator->GetTrace().Save(endInfo);
1167 1 : }
1168 :
1169 4 : return HCCL_SUCCESS;
1170 5 : }
1171 :
1172 8 : HcclResult HcclAllReduceV2(void *sendBuf, void *recvBuf, uint64_t count, HcclDataType dataType, HcclReduceOp op,
1173 : HcclComm comm, aclrtStream stream)
1174 : {
1175 8 : HcclUs startut = TIME_NOW();
1176 : bool isCapture;
1177 8 : rtModel_t rtModel = nullptr;
1178 8 : u32 modelId = 0;
1179 :
1180 8 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
1181 8 : const std::string tag = "AllReduce_" + communicator->GetId();
1182 :
1183 8 : CHK_RET_AND_PRINT_IDE(HcomCheckOpParamV2(tag.c_str(), count, dataType, stream), tag.c_str());
1184 8 : CHK_RET_AND_PRINT_IDE(HcomCheckReductionOpV2(op), tag.c_str());
1185 14 : CHK_RET_AND_PRINT_IDE(HcomCheckReduceDataTypeV2(dataType, op), tag.c_str());
1186 7 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
1187 :
1188 : /* 接口交互信息日志 */
1189 : char stackLogBufferV2[LOG_TMPBUF_SIZE];
1190 7 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1191 1 : s32 streamId = HrtGetStreamId(stream);
1192 1 : s32 deviceLogicId = HrtGetDevice();
1193 1 : u32 localRank = INVALID_VALUE_RANKID;
1194 1 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
1195 :
1196 2 : s32 ret = snprintf_s(stackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
1197 : "tag[%s], sendBuf[%p], recvBuf[%p], count[%llu], dataType[%s], op[%s], localRank[%u], streamId[%d],"
1198 : "comm[%p], deviceLogicId[%d]",
1199 3 : tag.c_str(), sendBuf, recvBuf, count, GetDataTypeEnumStrV2(dataType).c_str(), GetReduceOpEnumStrV2(op).c_str(),
1200 : localRank, streamId, comm, deviceLogicId);
1201 :
1202 1 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
1203 :
1204 1 : std::string logInfo = "Entry-HcclAllReduceV2: " + std::string(stackLogBufferV2);
1205 1 : if (isCapture) {
1206 0 : CHK_PTR_NULL(rtModel);
1207 : // 获取不到modelId会报错
1208 0 : CHK_RET(GetModelId(rtModel, modelId));
1209 0 : logInfo += ", model id[" + to_string(modelId) + "].";
1210 : }
1211 1 : communicator->GetTrace().Save(logInfo);
1212 1 : }
1213 :
1214 7 : static thread_local Hccl::CollOpParams opParams;
1215 7 : opParams.opType = Hccl::OpType::ALLREDUCE;
1216 7 : opParams.dataType = HcclDataTypeToDataType(dataType);
1217 7 : opParams.reduceOp = HCCL_OP_REDUCE_MAP[op];
1218 7 : opParams.sendBuf = sendBuf;
1219 7 : opParams.recvBuf = recvBuf;
1220 7 : opParams.count = count;
1221 7 : opParams.opTag = tag;
1222 7 : CHK_RET_AND_PRINT_IDE(communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream)), tag.c_str());
1223 :
1224 7 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1225 1 : HcclUs endut = TIME_NOW();
1226 : /* 关键状态记录 */
1227 1 : std::string endInfo = "HcclAllReduceV2:success,take time: " +
1228 2 : std::to_string(DURATION_US(endut - startut).count()) + " us," + std::string(stackLogBufferV2);
1229 1 : communicator->GetTrace().Save(endInfo);
1230 1 : }
1231 :
1232 7 : return HCCL_SUCCESS;
1233 8 : }
1234 :
1235 2 : HcclResult HcclBroadcastV2(void *buf, uint64_t count, HcclDataType dataType, uint32_t root, HcclComm comm,
1236 : aclrtStream stream)
1237 : {
1238 2 : HcclUs startut = TIME_NOW();
1239 : bool isCapture;
1240 2 : rtModel_t rtModel = nullptr;
1241 2 : u32 modelId = 0;
1242 :
1243 2 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
1244 2 : const std::string tag = "Broadcast_" + communicator->GetId();
1245 :
1246 2 : CHK_RET_AND_PRINT_IDE(HcomCheckOpParamV2(tag.c_str(), count, dataType, stream), tag.c_str());
1247 2 : u32 rankSize = INVALID_VALUE_RANKSIZE;
1248 2 : CHK_RET_AND_PRINT_IDE(communicator->GetRankSize(&rankSize), tag.c_str());
1249 2 : CHK_RET_AND_PRINT_IDE(HcomCheckUserRankV2(rankSize, root), tag.c_str());
1250 2 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
1251 :
1252 : /* 接口交互信息日志 */
1253 : char stackLogBufferV2[LOG_TMPBUF_SIZE];
1254 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1255 1 : s32 streamId = HrtGetStreamId(stream);
1256 1 : s32 deviceLogicId = HrtGetDevice();
1257 1 : u32 localRank = INVALID_VALUE_RANKID;
1258 1 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
1259 :
1260 1 : s32 ret = snprintf_s(stackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
1261 : "tag[%s], buf[%p], count[%llu], dataType[%s], root[%u], localRank[%u], streamId[%d], deviceLogicId[%d]",
1262 2 : tag.c_str(), buf, count, GetDataTypeEnumStrV2(dataType).c_str(), root, localRank, streamId, deviceLogicId);
1263 :
1264 1 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
1265 1 : std::string logInfo = "Entry-HcclBroadcastV2:" + std::string(stackLogBufferV2);
1266 1 : if (isCapture) {
1267 0 : CHK_PTR_NULL(rtModel);
1268 : // 获取不到modelId会报错
1269 0 : CHK_RET(GetModelId(rtModel, modelId));
1270 0 : logInfo += ", model id[" + to_string(modelId) + "].";
1271 : }
1272 1 : communicator->GetTrace().Save(logInfo);
1273 1 : }
1274 :
1275 2 : static thread_local Hccl::CollOpParams opParams;
1276 2 : opParams.opType = Hccl::OpType::BROADCAST;
1277 2 : opParams.dataType = HcclDataTypeToDataType(dataType);
1278 2 : opParams.sendBuf = buf;
1279 2 : opParams.recvBuf = buf;
1280 2 : opParams.count = count;
1281 2 : opParams.root = root;
1282 2 : opParams.opTag = tag;
1283 2 : CHK_RET_AND_PRINT_IDE(communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream)), tag.c_str());
1284 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1285 1 : HcclUs endut = TIME_NOW();
1286 : /* 关键状态记录 */
1287 1 : std::string endInfo = "HcclBroadcastV2:success,take time: " +
1288 2 : std::to_string(DURATION_US(endut - startut).count()) + " us," + std::string(stackLogBufferV2);
1289 1 : communicator->GetTrace().Save(endInfo);
1290 1 : }
1291 :
1292 2 : return HCCL_SUCCESS;
1293 2 : }
1294 :
1295 2 : HcclResult HcclBarrierV2(HcclComm comm, aclrtStream stream)
1296 : {
1297 2 : HcclUs startut = TIME_NOW();
1298 6 : HCCL_INFO("HcclBarrierV2 V82");
1299 : // AllReduce入参定义
1300 2 : HcclDataType dataType = HCCL_DATA_TYPE_FP32;
1301 2 : HcclReduceOp op = HCCL_REDUCE_SUM;
1302 2 : const uint64_t count = 8;
1303 2 : void *sendBuf = nullptr;
1304 2 : void *recvBuf = nullptr;
1305 2 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
1306 2 : s32 deviceLogicId = HcclGetThreadDeviceId();
1307 2 : s32 devPhyId = HrtGetDevicePhyIdByIndex(deviceLogicId);
1308 6 : HCCL_RUN_INFO("Entry-HcclBarrier V950, commId[%s], deviceLogicId[%d], devPhyId[%d]", communicator->GetId().c_str(),
1309 : deviceLogicId, devPhyId);
1310 : // 申请Device内存
1311 2 : auto ret = communicator->CreateBarrierMemory(sendBuf, recvBuf, count);
1312 2 : if (ret != HCCL_SUCCESS) {
1313 1 : return ret;
1314 : }
1315 : // 同通信域同算子复用tag
1316 1 : const string tag = "AllReduce_" + communicator->GetId();
1317 :
1318 1 : CHK_RET_AND_PRINT_IDE(HcomCheckOpParamV2(tag.c_str(), count, dataType, stream), tag.c_str());
1319 :
1320 1 : CHK_RET_AND_PRINT_IDE(HcomCheckReductionOpV2(op), tag.c_str());
1321 :
1322 1 : CHK_RET_AND_PRINT_IDE(HcomCheckReduceDataTypeV2(dataType, op), tag.c_str());
1323 :
1324 1 : static thread_local Hccl::CollOpParams opParams;
1325 1 : opParams.opType = Hccl::OpType::ALLREDUCE;
1326 1 : opParams.dataType = HcclDataTypeToDataType(dataType);
1327 1 : opParams.reduceOp = Hccl::ReduceOp::SUM;
1328 :
1329 1 : opParams.sendBuf = sendBuf;
1330 1 : opParams.recvBuf = recvBuf;
1331 1 : opParams.count = count;
1332 1 : opParams.opTag = tag;
1333 1 : ret = communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream));
1334 3 : HCCL_RUN_INFO("Entry-HcclBarrier V950 success, take time [%lld]us, commId[%s], deviceLogicId[%d]",
1335 : DURATION_US(TIME_NOW() - startut), communicator->GetId().c_str(), deviceLogicId);
1336 1 : return ret;
1337 1 : }
1338 :
1339 1 : HcclResult HcclGetHeterogModeV2(HcclComm comm, HcclHeterogMode *mode)
1340 : {
1341 : (void)comm;
1342 1 : *mode = HCCL_HETEROG_MODE_HOMOGENEOUS;
1343 3 : HCCL_INFO("[HcclGetHeterogModeV2] 950 only support homogeneous chip mode");
1344 1 : return HCCL_SUCCESS;
1345 : }
1346 :
1347 1 : HcclResult HcclCommSuspendV2(HcclComm comm)
1348 : {
1349 1 : CHK_PTR_NULL(comm);
1350 3 : HCCL_ERROR("HcclCommSuspend V950 not support suspend");
1351 :
1352 1 : return HCCL_E_NOT_SUPPORT;
1353 : }
1354 :
1355 6 : HcclResult HcclAllocComResourceByTilingV2(HcclComm comm, const void *stream, void *mc2Tiling, void **commContext)
1356 : {
1357 9 : CHK_PTR_NULL(comm);
1358 8 : CHK_PTR_NULL(stream);
1359 7 : CHK_PTR_NULL(mc2Tiling);
1360 6 : CHK_PTR_NULL(commContext);
1361 2 : HcclUs startut = TIME_NOW();
1362 2 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
1363 :
1364 2 : uint32_t *pVersion = reinterpret_cast<uint32_t *>(mc2Tiling);
1365 2 : u32 localRank = INVALID_VALUE_RANKID;
1366 : char stackLogBufferV2[LOG_TMPBUF_SIZE];
1367 2 : string commIdentifier = communicator->GetId();
1368 2 : const std::string tag = "HcclAllocComResourceByTilingV2_" + communicator->GetId();
1369 2 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
1370 : /* 接口交互信息日志 */
1371 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1372 1 : s32 ret = snprintf_s(stackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U, "commIdentifier[%s], version[%u]",
1373 : commIdentifier.c_str(), *pVersion);
1374 1 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, commIdentifier[%s].", commIdentifier.c_str()));
1375 :
1376 2 : std::string logInfo = "MC2 create resource by tiling: localRank[" + std::to_string(localRank)
1377 3 : + "]" + std::string(stackLogBufferV2);
1378 1 : communicator->GetTrace().Save(logInfo);
1379 1 : }
1380 :
1381 2 : HcclResult ret = communicator->AllocCommResource(mc2Tiling, commContext);
1382 2 : CHK_PRT_RET(ret != HCCL_SUCCESS,
1383 : HCCL_ERROR("[HcclAllocComResourceByTilingV2]AllocCommResource fail, errNo[%d], commIdentifier[%s]", ret,
1384 : commIdentifier.c_str()), ret);
1385 :
1386 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1387 1 : HcclUs endut = TIME_NOW();
1388 : /* 关键状态记录 */
1389 : std::string endInfo = "MC2 create resource take time ["
1390 2 : + std::to_string(DURATION_US(endut - startut).count()) + "]us, localRank["
1391 3 : + std::to_string(localRank) + "] " + std::string(stackLogBufferV2);
1392 1 : communicator->GetTrace().Save(endInfo);
1393 1 : }
1394 :
1395 2 : return HCCL_SUCCESS;
1396 2 : }
1397 :
1398 2 : HcclResult HcclGetOpArgsV2(void **opArgs)
1399 : {
1400 2 : CHK_PTR_NULL(opArgs);
1401 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1402 0 : HCCL_RUN_INFO("Entry-HcclGetOpArgs V950, start malloc opArgs in %p", opArgs);
1403 : }
1404 2 : HcclOpArgs *opArgsMem = (HcclOpArgs *)malloc(sizeof(HcclOpArgs));
1405 2 : if (opArgsMem == nullptr) {
1406 0 : HCCL_ERROR("[HcclGetOpArgs] malloc HcclOpArgs mem fail, please check.");
1407 0 : return HCCL_E_INTERNAL;
1408 : }
1409 2 : opArgsMem->Init();
1410 2 : *opArgs = opArgsMem;
1411 6 : HCCL_RUN_INFO("HcclGetOpArgs malloc HcclOpArgs success, please fill mem[%p->%p] in it.", opArgs, *opArgs);
1412 2 : return HCCL_SUCCESS;
1413 : }
1414 :
1415 2 : HcclResult HcclFreeOpArgsV2(void *opArgs)
1416 : {
1417 2 : CHK_PTR_NULL(opArgs);
1418 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1419 0 : HCCL_RUN_INFO("Entry-HcclFreeOpArgs V950, free opArgs[%p]", opArgs);
1420 : }
1421 2 : free(opArgs);
1422 2 : opArgs = nullptr;
1423 2 : return HCCL_SUCCESS;
1424 : }
1425 :
1426 2 : HcclResult HcclSetOpSrcDataTypeV2(void *opArgs, uint8_t srcDataType)
1427 : {
1428 2 : CHK_PTR_NULL(opArgs);
1429 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1430 0 : HCCL_RUN_INFO("Entry-HcclSetOpSrcDataType V950, opArgs[%p] set srcDataType[%u]", opArgs, srcDataType);
1431 : }
1432 2 : HcclOpArgs *opArgsPtr = static_cast<HcclOpArgs *>(opArgs);
1433 2 : if (srcDataType >= (sizeof(MC2_DATA_TYPE) / sizeof(MC2_DATA_TYPE[0]))) {
1434 3 : HCCL_ERROR("HcclSetOpSrcDataType set srcDataType[%u] error, it's invalid.", srcDataType);
1435 1 : return HCCL_E_PARA;
1436 : }
1437 1 : opArgsPtr->srcDataType = MC2_DATA_TYPE[srcDataType];
1438 1 : return HCCL_SUCCESS;
1439 : }
1440 :
1441 2 : HcclResult HcclSetOpDstDataTypeV2(void *opArgs, uint8_t dstDataType)
1442 : {
1443 2 : CHK_PTR_NULL(opArgs);
1444 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1445 0 : HCCL_RUN_INFO("Entry-HcclSetOpDstDataType V950, opArgs[%p] set dstDataType[%u]", opArgs, dstDataType);
1446 : }
1447 2 : HcclOpArgs *opArgsPtr = static_cast<HcclOpArgs *>(opArgs);
1448 2 : if (dstDataType >= (sizeof(MC2_DATA_TYPE) / sizeof(MC2_DATA_TYPE[0]))) {
1449 3 : HCCL_ERROR("HcclSetOpDstDataType set dstDataType[%u] error, it's invalid.", dstDataType);
1450 1 : return HCCL_E_PARA;
1451 : }
1452 1 : opArgsPtr->dstDataType = MC2_DATA_TYPE[dstDataType];
1453 1 : return HCCL_SUCCESS;
1454 : }
1455 :
1456 2 : HcclResult HcclSetOpReduceTypeV2(void *opArgs, uint32_t reduceType)
1457 : {
1458 2 : CHK_PTR_NULL(opArgs);
1459 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1460 0 : HCCL_RUN_INFO("Entry-HcclSetOpReduceType V950, opArgs[%p] set reduceType[%u]", opArgs, reduceType);
1461 : }
1462 2 : HcclOpArgs *opArgsPtr = static_cast<HcclOpArgs *>(opArgs);
1463 2 : if (reduceType >= (sizeof(MC2_REDUCE_TYPE) / sizeof(MC2_REDUCE_TYPE[0]))) {
1464 3 : HCCL_ERROR("HcclSetOpReduceType set reduceType[%u] error, it's invalid.", reduceType);
1465 1 : return HCCL_E_PARA;
1466 : }
1467 1 : opArgsPtr->reduceType = MC2_REDUCE_TYPE[reduceType];
1468 1 : return HCCL_SUCCESS;
1469 : }
1470 :
1471 2 : HcclResult HcclSetOpCountV2(void *opArgs, uint64_t count)
1472 : {
1473 2 : CHK_PTR_NULL(opArgs);
1474 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1475 0 : HCCL_RUN_INFO("Entry-HcclSetOpCount V950, opArgs[%p] set count[%llu]", opArgs, count);
1476 : }
1477 2 : HcclOpArgs *opArgsPtr = static_cast<HcclOpArgs *>(opArgs);
1478 5 : CHK_RET(HcomCheckCountV2(count));
1479 1 : opArgsPtr->count = count;
1480 1 : return HCCL_SUCCESS;
1481 : }
1482 :
1483 2 : HcclResult HcclSetOpAlgConfigV2(void *opArgs, char *algConfig)
1484 : {
1485 2 : CHK_PTR_NULL(opArgs);
1486 2 : CHK_PTR_NULL(algConfig);
1487 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1488 0 : HCCL_RUN_INFO("Entry-HcclSetOpAlgConfig V950, opArgs[%p]", opArgs);
1489 : }
1490 2 : HcclOpArgs *opArgsPtr = static_cast<HcclOpArgs *>(opArgs);
1491 2 : s32 ret = strcpy_s(opArgsPtr->algConfig, ALG_CONFIG_SIZE, algConfig);
1492 2 : if (ret != EOK) {
1493 3 : HCCL_ERROR("[HcclSetOpAlgConfig]strcpy_s algConfig failed! result %d, the algConfig len must be less than %u", ret, ALG_CONFIG_SIZE);
1494 1 : return HCCL_E_PARA;
1495 : }
1496 1 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1497 0 : HCCL_RUN_INFO("HcclSetOpAlgConfig set opArgs[%p] algConfig success, algConfig is [%s]", opArgs, opArgsPtr->algConfig);
1498 : }
1499 1 : return HCCL_SUCCESS;
1500 : };
1501 :
1502 1 : HcclResult HcclSetOpCommEngineV2(void *opArgs, uint8_t commEngine)
1503 : {
1504 1 : CHK_PTR_NULL(opArgs);
1505 1 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1506 0 : HCCL_RUN_INFO("Entry-HcclSetOpCommEngine V950, commEngine[%u]", commEngine);
1507 : }
1508 1 : HcclOpArgs *opArgsPtr = static_cast<HcclOpArgs *>(opArgs);
1509 1 : opArgsPtr->commEngine = HcclAccelerator(static_cast<HcclAccelerator::Value>(commEngine));
1510 1 : return HCCL_SUCCESS;
1511 : }
1512 :
1513 1 : HcclResult HcclCommResPrepareWithOpMode(Hccl::HcclCommunicator *communicator, const std::string &opName, HcclOpArgs *opArgs, void **addr)
1514 : {
1515 1 : CHK_PTR_NULL(communicator);
1516 1 : CHK_PTR_NULL(opArgs);
1517 1 : CHK_PTR_NULL(addr);
1518 1 : OpType opType = GetOpTypeV2(opName);
1519 1 : if (opType == OpType::OPTYPEINVALID) {
1520 0 : HCCL_ERROR("[HcclCommResPrepareWithOpMode] The opName %s match opType is %s", opName.c_str(), opType.Describe().c_str());
1521 0 : return HCCL_E_PARA;
1522 : }
1523 :
1524 1 : std::string opTag = opName + communicator->GetId() + "_mc2";
1525 1 : static thread_local Hccl::CollOpParams opParams;
1526 1 : opParams.opType = opType;
1527 1 : opParams.reduceOp = opArgs->reduceType;
1528 1 : opParams.dataType = opArgs->srcDataType;
1529 1 : opParams.outputDataType = opArgs->dstDataType;
1530 1 : opParams.count = opArgs->count;
1531 1 : opParams.opTag = opTag;
1532 1 : opParams.algConfig = std::string(opArgs->algConfig);
1533 1 : opParams.isMc2 = true;
1534 1 : opParams.commEngine = opArgs->commEngine;
1535 1 : return communicator->AllocCollOpResource(opParams, addr);
1536 1 : }
1537 :
1538 1 : HcclResult HcclCommResPrepareV2(HcclComm comm, char *opName, void* opArgs, void **addr)
1539 : {
1540 1 : CHK_PTR_NULL(comm);
1541 1 : CHK_PTR_NULL(opName);
1542 1 : CHK_PTR_NULL(opArgs);
1543 1 : CHK_PTR_NULL(addr);
1544 1 : std::string opNameStr(opName);
1545 1 : opNameStr = opNameStr.substr(0, MAX_OP_NAME_SIZE);
1546 1 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
1547 1 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1548 0 : HCCL_RUN_INFO("Entry-HcclCommResPrepare V950, opName[%s], opArgs addr[%p], commId[%s]", opNameStr.c_str(), opArgs, communicator->GetId().c_str());
1549 : }
1550 1 : return HcclCommResPrepareWithOpMode(communicator, opNameStr, static_cast<HcclOpArgs *>(opArgs), addr);
1551 1 : }
1552 :
1553 2 : HcclResult HcclDevMemAcquireV2(HcclComm comm, const char *memTag, uint64_t *size, void **addr, bool *newCreated)
1554 : {
1555 2 : CHK_PTR_NULL(comm);
1556 2 : CHK_PTR_NULL(memTag);
1557 2 : CHK_PTR_NULL(size);
1558 2 : CHK_PTR_NULL(addr);
1559 2 : CHK_PTR_NULL(newCreated);
1560 2 : std::string memTagStr = "";
1561 2 : if (memTag != nullptr) {
1562 : char tmpMemTag[MAX_MEM_TAG_SIZE];
1563 2 : s32 ret = strcpy_s(tmpMemTag, MAX_MEM_TAG_SIZE, memTag);
1564 2 : if (ret != EOK) {
1565 0 : HCCL_ERROR("[HcclDevMemAcquire] strcpy_s memTag failed! result %d, the memTag len must be less than %u", ret, MAX_MEM_TAG_SIZE);
1566 0 : return HCCL_E_PARA;
1567 : }
1568 4 : memTagStr = std::string(tmpMemTag);
1569 : }
1570 2 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
1571 2 : CHK_RET(communicator->GetDevMemWorkSpace(memTagStr, size, addr, newCreated));
1572 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1573 0 : HCCL_RUN_INFO("Entry-HcclDevMemAcquire V950, memTag[%s], addr[%p], size[%llu], commId[%s]", memTagStr.c_str(), *addr, *size, communicator->GetId().c_str());
1574 : }
1575 2 : return HCCL_SUCCESS;
1576 2 : }
1577 :
1578 1 : HcclResult HcclGetHcclBufferV2(HcclComm comm, void **addr, uint64_t *size)
1579 : {
1580 1 : CHK_PTR_NULL(comm);
1581 1 : CHK_PTR_NULL(addr);
1582 1 : CHK_PTR_NULL(size);
1583 1 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
1584 1 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1585 0 : HCCL_RUN_INFO("Entry-HcclGetHcclBuffer V950, commId[%s]", communicator->GetId().c_str());
1586 : }
1587 1 : CHK_RET(communicator->GetLocalCclBuffer(addr, size));
1588 : /* 关键状态记录 */
1589 1 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1590 0 : HCCL_RUN_INFO("HcclGetHcclBuffer success, addr[%p], size[%llu], commId[%s]", addr, *size, communicator->GetId().c_str());
1591 : }
1592 1 : return HCCL_SUCCESS;
1593 : }
1594 :
1595 1 : HcclResult HcclGetRemoteIpcHcclBufV2(HcclComm comm, uint64_t remoteRank, void **addr, uint64_t *size)
1596 : {
1597 1 : CHK_PTR_NULL(comm);
1598 1 : CHK_PTR_NULL(addr);
1599 1 : CHK_PTR_NULL(size);
1600 1 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
1601 1 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1602 0 : HCCL_RUN_INFO("Entry-HcclGetRemoteIpcHcclBuf V950 start, remoteRank[%llu], addr[%p], size[%llu], commId[%s]", remoteRank, *addr, *size, communicator->GetId().c_str());
1603 : }
1604 3 : HCCL_ERROR("Entry-HcclGetRemoteIpcHcclBuf V950 not support, commId[%s]", communicator->GetId().c_str());
1605 1 : return HCCL_E_NOT_SUPPORT;
1606 : }
1607 :
1608 1 : HcclResult HcclGetAicpuOpStreamAndNotifyV2(HcclComm comm, rtStream_t *opstream, u8 aicpuNotifyNum, void **aicpuNotify)
1609 : {
1610 1 : CHK_PTR_NULL(comm);
1611 1 : CHK_PTR_NULL(opstream);
1612 1 : CHK_PTR_NULL(aicpuNotify);
1613 1 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
1614 1 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1615 0 : HCCL_RUN_INFO("Entry-HcclGetAicpuOpStreamAndNotify V950, aicpuNotifyNum[%u], commId[%s]", aicpuNotifyNum, communicator->GetId().c_str());
1616 : }
1617 1 : CHK_RET(communicator->GetAicpuOpStreamNotify(opstream, aicpuNotifyNum, aicpuNotify));
1618 1 : return HCCL_SUCCESS;
1619 : }
1620 :
1621 2 : HcclResult HcclCommResumeV2(HcclComm comm)
1622 : {
1623 2 : CHK_PTR_NULL(comm);
1624 2 : HcclUs startut = TIME_NOW();
1625 6 : HCCL_RUN_INFO("Entry-HcclCommResume V950");
1626 2 : s32 deviceLogicId = HcclGetThreadDeviceId();
1627 2 : s32 devPhyId = HrtGetDevicePhyIdByIndex(deviceLogicId);
1628 2 : CHK_RET(static_cast<HcclResult>(Hccl::HcclCcuResumePfeTableProcess(deviceLogicId)));
1629 2 : CHK_RET(HcclCommResumeImplV2(comm));
1630 6 : HCCL_RUN_INFO("Entry-HcclCommResume V950 success, deviceLogicId[%d], devPhyId[%d], take time [%lld]us",
1631 : deviceLogicId, devPhyId, DURATION_US(TIME_NOW() - startut));
1632 2 : return HCCL_SUCCESS;
1633 : }
1634 :
1635 2 : HcclResult HcclCommResumeImplV2(HcclComm comm)
1636 : {
1637 2 : CHK_PTR_NULL(comm);
1638 6 : return HcclCommOperationImplV2(comm, "HcclCommResumeV2", [](Hccl::HcclCommunicator &communicator) {
1639 6 : HCCL_RUN_INFO("Entry-HcclCommResume commId[%s]", communicator.GetId().c_str());
1640 2 : return static_cast<HcclResult>(communicator.Resume());
1641 2 : });
1642 : }
1643 :
1644 5 : HcclResult RootInfoDetect(std::shared_ptr<RankInfoDetect> rankInfoDetectAgent, const u32 nRanks, u32 rank,
1645 : const HcclRootHandleV2 &rootHandle, RankTableInfo &rankTable)
1646 : {
1647 5 : s32 deviceLogicId = HcclGetThreadDeviceId();
1648 5 : s32 devPhyId = HrtGetDevicePhyIdByIndex(deviceLogicId);
1649 15 : HCCL_RUN_INFO("[%s] nRanks[%u], rank[%u] entry flat topo detect, rootinfo: host ip[%s] port[%u] netMode[%s] "
1650 : "identifier[%s], deviceLogicId[%d], devPhyId[%d]",
1651 : __func__, nRanks, rank, rootHandle.ip, rootHandle.listenPort,
1652 : rootHandle.netMode.Describe().c_str(), rootHandle.identifier, deviceLogicId, devPhyId);
1653 : // client端拓扑探测
1654 :
1655 5 : bool hasException = false;
1656 8 : EXCEPTION_CATCH(rankInfoDetectAgent->SetupAgent(nRanks, rank, rootHandle), hasException = true);
1657 :
1658 : // 等server端执行结束
1659 5 : EXCEPTION_CATCH(rankInfoDetectAgent->WaitComplete(rootHandle.listenPort, RANKINFO_DETECT_SERVER_STATUS_IDLE), hasException = true);
1660 :
1661 : // 若探测流程异常返回错误信息
1662 8 : CHK_PRT_RET(hasException, HCCL_ERROR("[%s] RankInfoDetect SetupAgent fail, identifier[%s].", __func__, rootHandle.identifier), HCCL_E_INTERNAL);
1663 :
1664 : // server 探测已结束（或失败），释放 GetRootInfo 保活的 server 对象，析构中 join 子线程
1665 4 : HcclCommInfoV2& opbasedCommInfoV2 = GetCommInfoV2();
1666 : {
1667 4 : std::lock_guard<std::mutex> detectServerGuard(opbasedCommInfoV2.detectServerLock);
1668 8 : auto iterServer = opbasedCommInfoV2.hcclCommRankInfoDetectServer.find(rootHandle.identifier);
1669 4 : if (iterServer != opbasedCommInfoV2.hcclCommRankInfoDetectServer.end()) {
1670 2 : opbasedCommInfoV2.hcclCommRankInfoDetectServer.erase(iterServer);
1671 6 : HCCL_INFO("[%s] release RankInfoDetect server, identifier[%s]", __func__, rootHandle.identifier);
1672 : }
1673 4 : }
1674 :
1675 : // 获取ranktable
1676 4 : rankInfoDetectAgent->GetRankTable(rankTable);
1677 :
1678 12 : HCCL_RUN_INFO("[%s] end.", __func__);
1679 4 : return HCCL_SUCCESS;
1680 : }
1681 :
1682 3 : HcclResult CommInitRootInfo(u32 nRanks, u32 rank, const HcclRootHandleV2 &rootHandle,
1683 : const string &identifier, HcclComm *comm)
1684 : {
1685 : // 临时规避，在初始化通信域前声明单例保证时序
1686 3 : CHK_RET(CallSingletons());
1687 : // check
1688 3 : HcclCommInfoV2 &opbasedCommInfoV2 = GetCommInfoV2();
1689 6 : CHK_PRT_RET(opbasedCommInfoV2.hcclGroupMap.find(identifier) != opbasedCommInfoV2.hcclGroupMap.end(),
1690 : HCCL_ERROR("[CreateCommConfig] errNo[0x%016llx] The rootHandle[%s] already exists in Group2Comm map.",
1691 : HCCL_ERROR_CODE(HCCL_E_PARA), identifier.c_str()), HCCL_E_PARA);
1692 :
1693 : // rootInfo获取rankTable, 基于rankTable创建通信域
1694 2 : std::shared_ptr<RankInfoDetect> rankInfoDetectAgent = std::make_shared<RankInfoDetect>();
1695 2 : RankTableInfo rankTable{};
1696 2 : HcclResult ret = RootInfoDetect(rankInfoDetectAgent, nRanks, rank, rootHandle, rankTable);
1697 2 : if (ret != HCCL_SUCCESS) {
1698 0 : RPT_INPUT_ERR(true, "EI0015", std::vector<std::string>({"error_reason"}),
1699 : std::vector<std::string>({"RootInfoDetect failed"}));
1700 0 : HCCL_ERROR("[%s] errNo[0x%016llx] RootInfoDetect failed, rootHandle[%s].", __func__, HCCL_ERROR_CODE(ret), identifier.c_str());
1701 0 : rankTable.Dump();
1702 0 : return ret;
1703 : }
1704 :
1705 : // 打印ranktable
1706 2 : rankTable.Dump();
1707 :
1708 : // 创建通信域
1709 2 : bool devUsed = false;
1710 2 : bool isWorldGroup = true;
1711 : Hccl::CommParams commParams{identifier, static_cast<Hccl::RankId>(rank), nRanks,
1712 2 : static_cast<Hccl::RankId>(rank), Hccl::HrtGetDeviceType(), devUsed, isWorldGroup};
1713 2 : opbasedCommInfoV2.pComm.reset(new (std::nothrow) Hccl::HcclCommunicator(commParams));
1714 2 : opbasedCommInfoV2.commParams = commParams;
1715 :
1716 : // 通信域初始化
1717 2 : CHK_PTR_NULL(opbasedCommInfoV2.pComm);
1718 : /* --------------初始化------------------------- */
1719 2 : bool errorFlag = false;
1720 2 : s32 logicDevId = HrtGetDevice();
1721 : do {
1722 2 : ret = opbasedCommInfoV2.pComm->Init(rankTable);
1723 2 : CHK_PRT_BREAK(ret != HcclResult::HCCL_SUCCESS,
1724 : HCCL_ERROR("[%s]opbasedCommInfoV2.pComm->Init failed, errNo[0x%016llx]", __func__, HCCL_ERROR_CODE(ret)),
1725 : errorFlag = true);
1726 : // 配置默认加速模式
1727 2 : opbasedCommInfoV2.pComm->RegisterAcceStateCallBack(CommunicatorCallback());
1728 2 : ret = CommManager::GetInstance(logicDevId).SetCommAcceleratorV2(opbasedCommInfoV2.pComm.get(), 0); // 通信域创建，设置默认accelerator
1729 2 : CHK_PRT_BREAK(ret != HcclResult::HCCL_SUCCESS,
1730 : HCCL_ERROR("[%s]SetCommAcceleratorV2 failed, errNo[0x%016llx]", __func__, HCCL_ERROR_CODE(ret)),
1731 : errorFlag = true);
1732 : // 保存通信域
1733 2 : HcclGroupParamsV2 params{};
1734 2 : params.pComm = opbasedCommInfoV2.pComm;
1735 2 : params.groupRank = static_cast<Hccl::RankId>(rank);
1736 2 : std::unique_lock<std::mutex> lock(opbasedCommInfoV2.groupParamsLock);
1737 2 : opbasedCommInfoV2.hcclGroupMap[identifier] = params;
1738 :
1739 4 : opbasedCommInfoV2.pComm->RegisterPrintChannelInfoCallback(
1740 4 : CommManager::GetInstance(logicDevId).GetPrintChannelInfoCallback());
1741 :
1742 2 : *comm = static_cast<HcclComm>(opbasedCommInfoV2.pComm.get());
1743 2 : } while (0);
1744 :
1745 2 : if (errorFlag) {
1746 0 : HCCL_ERROR("[Init][%s]HcclCommInitClusterInfoV2 failed, rankNum[%u], rank[%u], logicDevId[%d], rootInfo identifier[%s],"\
1747 : "return[0x%016llx]", __func__, nRanks, rank,
1748 : logicDevId, identifier.c_str(), HCCL_ERROR_CODE(ret));
1749 0 : (void)HcclCommDestroyV2(opbasedCommInfoV2.pComm.get());
1750 0 : *comm = nullptr;
1751 0 : return ret;
1752 : }
1753 6 : HCCL_INFO("[%s] Init success, rankNum[%u], rank[%u], rootInfo identifier[%s], logicDevId[%d]", __func__,
1754 : nRanks, rank, identifier.c_str(), logicDevId);
1755 :
1756 2 : return HCCL_SUCCESS;
1757 2 : }
1758 :
1759 3 : HcclResult HcclCommInitRootInfoV2(
1760 : uint32_t nRanks, const HcclRootInfo *rootInfo, uint32_t rank, HcclComm *comm, std::string &identifier)
1761 : {
1762 3 : HcclUs startut = TIME_NOW();
1763 3 : CHK_PTR_NULL(rootInfo);
1764 9 : HCCL_RUN_INFO("Entry-HcclCommInitRootInfo V950, rankId[%u], rankNum[%u].", rank, nRanks);
1765 :
1766 : // 获取rootHandle
1767 3 : HcclRootHandleV2 rootHandle{};
1768 3 : s32 sRet = memcpy_s(&rootHandle, sizeof(HcclRootHandleV2), rootInfo->internal, sizeof(HcclRootHandleV2));
1769 3 : CHK_PRT_RET(sRet != EOK, HCCL_ERROR("[%s] memcpy root info fail. errorno[%d] length[%u]", __func__,
1770 : sRet, sizeof(rootHandle)), HCCL_E_MEMORY);
1771 :
1772 : // 获取通信域name
1773 3 : rootHandle.identifier[ROOTINFO_INDENTIFIER_MAX_LENGTH - 1] = '\0';
1774 3 : identifier = rootHandle.identifier;
1775 :
1776 : /* 接口交互信息日志 */
1777 3 : s32 deviceLogicId = HcclGetThreadDeviceId();
1778 3 : s32 devPhyId = HrtGetDevicePhyIdByIndex(deviceLogicId);
1779 9 : HCCL_RUN_INFO("Entry-HcclCommInitRootInfo V950, ranks[%u], rank[%u], rootinfo: host ip[%s] port[%u] "\
1780 : "netMode[%s] identifier[%s], deviceLogicId[%d], devPhyId[%d]", nRanks, rank, rootHandle.ip, rootHandle.listenPort,
1781 : rootHandle.netMode.Describe().c_str(), identifier.c_str(), deviceLogicId, devPhyId);
1782 :
1783 : // rootInfo获取rankTable, 基于rankTable创建通信域
1784 3 : HcclResult ret = CommInitRootInfo(nRanks, rank, rootHandle, identifier, comm);
1785 6 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[%s] errNo[0x%016llx] CommInitRootInfo failed, identifier[%s].",
1786 : __func__, HCCL_ERROR_CODE(ret), identifier.c_str()), ret);
1787 :
1788 : /* 关键状态记录 */
1789 6 : HCCL_RUN_INFO("HcclCommInitRootInfoV2 success, take time [%lld]us, rankNum[%u], rank[%u], identifier[%s]",
1790 : DURATION_US(TIME_NOW() - startut), nRanks, rank, identifier.c_str());
1791 2 : return HCCL_SUCCESS;
1792 : }
1793 :
1794 3 : HcclResult HcclCommInitRootInfoConfigV2(uint32_t nRanks, const HcclRootInfo *rootInfo, uint32_t rank,
1795 : const HcclCommConfig *config, HcclComm *comm)
1796 : {
1797 3 : HcclUs startut = TIME_NOW();
1798 3 : CHK_PTR_NULL(rootInfo);
1799 3 : CHK_PTR_NULL(config);
1800 9 : HCCL_RUN_INFO("Entry-HcclCommInitRootInfoConfig V950: nRanks[%u], rank[%u], commEngine[%u]", nRanks, rank, config->hcclOpExpansionMode);
1801 : // 获取rootHandle
1802 3 : HcclRootHandleV2 rootHandle{};
1803 3 : s32 sRet = memcpy_s(&rootHandle, sizeof(rootHandle), rootInfo->internal, sizeof(rootHandle));
1804 3 : CHK_PRT_RET(sRet != EOK, HCCL_ERROR("[%s]memcpy root info fail. errorno[%d] count[%u]", __func__,
1805 : sRet, sizeof(HcclRootHandleV2)), HCCL_E_MEMORY);
1806 :
1807 : // 获取通信域name
1808 3 : rootHandle.identifier[ROOTINFO_INDENTIFIER_MAX_LENGTH - 1] = '\0';
1809 3 : string identifier = strlen(config->hcclCommName) != 0 ? config->hcclCommName : rootHandle.identifier;
1810 :
1811 : /* 接口交互信息日志 */
1812 9 : HCCL_RUN_INFO("Entry-HcclCommInitRootInfoConfigV2:ranks[%u], rank[%u], rootinfo: host ip[%s] port[%u] "
1813 : "netMode[%s] rootHandle.identifier[%s], identifier[%s]", nRanks, rank, rootHandle.ip, rootHandle.listenPort,
1814 : rootHandle.netMode.Describe().c_str(), rootHandle.identifier, identifier.c_str());
1815 :
1816 : // 临时规避，在初始化通信域前声明单例保证时序
1817 3 : CHK_RET(CallSingletons());
1818 3 : HcclCommInfoV2 &opbasedCommInfoV2 = GetCommInfoV2();
1819 3 : CHK_PRT_RET(opbasedCommInfoV2.hcclGroupMap.find(identifier) != opbasedCommInfoV2.hcclGroupMap.end(),
1820 : HCCL_ERROR("[HcclCommInitRootInfoConfigV2]errNo[0x%016llx] The comm name[%s] already exists in Group2Comm map.",
1821 : HCCL_ERROR_CODE(HCCL_E_PARA), identifier.c_str()), HCCL_E_PARA);
1822 :
1823 3 : RankTableInfo rankTable{};
1824 3 : std::shared_ptr<RankInfoDetect> rankInfoDetectAgent = std::make_shared<RankInfoDetect>();
1825 3 : HcclResult ret = RootInfoDetect(rankInfoDetectAgent, nRanks, rank, rootHandle, rankTable);
1826 3 : if (ret != HCCL_SUCCESS) {
1827 1 : RPT_INPUT_ERR(true, "EI0015", std::vector<std::string>({"error_reason"}),
1828 : std::vector<std::string>({"RootInfoDetect failed"}));
1829 3 : HCCL_ERROR("[%s] errNo[0x%016llx] RootInfoDetect failed, identifier[%s].", __func__, HCCL_ERROR_CODE(ret), identifier.c_str());
1830 1 : rankTable.Dump();
1831 1 : return ret;
1832 : }
1833 :
1834 : // 打印ranktable
1835 2 : rankTable.Dump();
1836 :
1837 : // 创建通信域
1838 2 : ret = CreateCommConfigRootInfo(rank, config, identifier, rankTable, comm);
1839 2 : CHK_PRT_RET(ret, HCCL_ERROR("[%s]errNo[0x%016llx] and create comm failed, identifier[%s].", __func__,
1840 : HCCL_ERROR_CODE(ret), identifier.c_str()), static_cast<HcclResult>(ret));
1841 :
1842 : /* 关键状态记录 */
1843 6 : HCCL_RUN_INFO("HcclCommInitRootInfoConfigV2 success, take time [%lld]us, rankNum[%u], rank[%u], identifier[%s]",
1844 : DURATION_US(TIME_NOW() - startut), nRanks, rank, identifier.c_str());
1845 2 : return HCCL_SUCCESS;
1846 3 : }
1847 :
1848 2 : HcclResult HcclScatterV2(void *sendBuf, void *recvBuf, uint64_t recvCount, HcclDataType dataType, uint32_t root,
1849 : HcclComm comm, aclrtStream stream)
1850 : {
1851 2 : HcclUs startut = TIME_NOW();
1852 : bool isCapture;
1853 2 : rtModel_t rtModel = nullptr;
1854 2 : u32 modelId = 0;
1855 :
1856 2 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
1857 2 : const std::string tag = "Scatter_" + communicator->GetId();
1858 :
1859 2 : CHK_RET_AND_PRINT_IDE(HcomCheckOpParamV2(tag.c_str(), recvCount, dataType, stream), tag.c_str());
1860 2 : u32 rankSize = INVALID_VALUE_RANKSIZE;
1861 2 : CHK_RET_AND_PRINT_IDE(communicator->GetRankSize(&rankSize), tag.c_str());
1862 2 : CHK_RET_AND_PRINT_IDE(HcomCheckUserRankV2(rankSize, root), tag.c_str());
1863 2 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
1864 :
1865 2 : u32 rankId = INVALID_VALUE_RANKID;
1866 2 : CHK_RET(communicator->GetRankId(rankId));
1867 2 : if (rankId == root) { // 本rank为root节点，send_buff不为空
1868 2 : CHK_PTR_NULL(sendBuf);
1869 : }
1870 :
1871 : /* 接口交互信息日志 */
1872 : char stackLogBufferV2[LOG_TMPBUF_SIZE];
1873 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1874 1 : s32 streamId = HrtGetStreamId(stream);
1875 1 : s32 deviceLogicId = HrtGetDevice();
1876 1 : u32 localRank = INVALID_VALUE_RANKID;
1877 1 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
1878 :
1879 1 : s32 ret = snprintf_s(stackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
1880 : "tag[%s], sendBuf[%p], recvBuf[%p], recvCount[%llu], dataType[%s], root[%u], localRank[%u], streamId[%d], deviceLogicId[%d]",
1881 2 : tag.c_str(), sendBuf, recvBuf, recvCount, GetDataTypeEnumStrV2(dataType).c_str(), root, localRank, streamId, deviceLogicId);
1882 :
1883 1 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
1884 1 : std::string logInfo = "Entry-HcclScatterV2:" + std::string(stackLogBufferV2);
1885 1 : if (isCapture) {
1886 0 : CHK_PTR_NULL(rtModel);
1887 : // 获取不到modelId会报错
1888 0 : CHK_RET(GetModelId(rtModel, modelId));
1889 0 : logInfo += ", model id[" + to_string(modelId) + "].";
1890 : }
1891 1 : communicator->GetTrace().Save(logInfo);
1892 1 : }
1893 :
1894 2 : static thread_local Hccl::CollOpParams opParams;
1895 2 : opParams.opType = Hccl::OpType::SCATTER;
1896 2 : opParams.dataType = HcclDataTypeToDataType(dataType);
1897 2 : opParams.sendBuf = sendBuf;
1898 2 : opParams.recvBuf = recvBuf;
1899 2 : opParams.count = recvCount;
1900 2 : opParams.root = root;
1901 :
1902 2 : CHK_RET_AND_PRINT_IDE(communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream)), tag.c_str());
1903 :
1904 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1905 1 : HcclUs endut = TIME_NOW();
1906 : /* 关键状态记录 */
1907 1 : std::string endInfo = "HcclScatterV2:success,take time: " +
1908 2 : std::to_string(DURATION_US(endut - startut).count()) + " us, tag: " + tag + std::string(stackLogBufferV2);
1909 1 : communicator->GetTrace().Save(endInfo);
1910 1 : }
1911 :
1912 2 : return HCCL_SUCCESS;
1913 2 : }
1914 :
1915 3 : HcclResult HcclAllGatherV2(void *sendBuf, void *recvBuf, uint64_t sendCount, HcclDataType dataType,
1916 : HcclComm comm, aclrtStream stream)
1917 : {
1918 3 : HcclUs startut = TIME_NOW();
1919 : bool isCapture;
1920 3 : rtModel_t rtModel = nullptr;
1921 3 : u32 modelId = 0;
1922 :
1923 3 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
1924 3 : const std::string tag = "AllGather_" + communicator->GetId();
1925 :
1926 3 : CHK_RET_AND_PRINT_IDE(HcomCheckOpParamV2(tag.c_str(), sendCount, dataType, stream), tag.c_str());
1927 3 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
1928 :
1929 : /* 接口交互信息日志 */
1930 : char stackLogBufferV2[LOG_TMPBUF_SIZE];
1931 3 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1932 1 : s32 streamId = HrtGetStreamId(stream);
1933 1 : s32 deviceLogicId = HrtGetDevice();
1934 1 : u32 localRank = INVALID_VALUE_RANKID;
1935 1 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
1936 :
1937 1 : s32 ret = snprintf_s(stackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
1938 : "tag[%s], sendBuf[%p], recvBuf[%p], sendCount[%llu], dataType[%s], localRank[%u], streamId[%d],"
1939 : "deviceLogicId[%d]",
1940 2 : tag.c_str(), sendBuf, recvBuf, sendCount, GetDataTypeEnumStrV2(dataType).c_str(), localRank, streamId, deviceLogicId);
1941 :
1942 1 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
1943 1 : std::string logInfo = "Entry-HcclAllGatherV2:" + std::string(stackLogBufferV2);
1944 1 : if (isCapture) {
1945 0 : CHK_PTR_NULL(rtModel);
1946 : // 获取不到modelId会报错
1947 0 : CHK_RET(GetModelId(rtModel, modelId));
1948 0 : logInfo += ", model id[" + to_string(modelId) + "].";
1949 : }
1950 1 : communicator->GetTrace().Save(logInfo);
1951 1 : }
1952 :
1953 3 : static thread_local Hccl::CollOpParams opParams;
1954 3 : opParams.opType = Hccl::OpType::ALLGATHER;
1955 3 : opParams.dataType = HcclDataTypeToDataType(dataType);
1956 3 : opParams.sendBuf = sendBuf;
1957 3 : opParams.recvBuf = recvBuf;
1958 3 : opParams.count = sendCount;
1959 3 : opParams.opTag = tag;
1960 :
1961 3 : CHK_RET_AND_PRINT_IDE(communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream)), tag.c_str());
1962 :
1963 3 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1964 1 : HcclUs endut = TIME_NOW();
1965 : /* 关键状态记录 */
1966 1 : std::string endInfo = "HcclAllGatherV2:success,take time: " +
1967 2 : std::to_string(DURATION_US(endut - startut).count()) + " us, tag: " + tag + std::string(stackLogBufferV2);
1968 1 : communicator->GetTrace().Save(endInfo);
1969 1 : }
1970 :
1971 3 : return HCCL_SUCCESS;
1972 3 : }
1973 :
1974 9 : HcclResult HcclAllGatherVV2(void *sendBuf, uint64_t sendCount, void *recvBuf, void *recvCounts, void *recvDispls,
1975 : HcclDataType dataType, HcclComm comm, aclrtStream stream)
1976 : {
1977 9 : HcclUs startut = TIME_NOW();
1978 : bool isCapture;
1979 9 : rtModel_t rtModel = nullptr;
1980 9 : u32 modelId = 0;
1981 :
1982 : // 获取通信域
1983 9 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
1984 9 : const std::string tag = "AllGatherV_" + communicator->GetId();
1985 :
1986 : // 获取rank信息
1987 : uint32_t rankId;
1988 9 : CHK_RET(communicator->GetRankId(rankId));
1989 : uint32_t rankSize;
1990 9 : CHK_RET(communicator->GetRankSize(&rankSize));
1991 : // 参数合法性校验
1992 9 : if (rankSize == 1) {
1993 : // rankSize为1时，退化为AllGather
1994 : // 检查异常回退AGV的情况
1995 3 : if (sendCount == 0) {
1996 3 : HCCL_WARNING("[AllGatherV] sendCount is 0 when single rank");
1997 1 : return HCCL_SUCCESS;
1998 : } else {
1999 5 : CHK_PRT_RET(sendBuf == nullptr, HCCL_ERROR("[AllGatherV] sendBuf is null when single rank"), HCCL_E_PTR);
2000 1 : CHK_PRT_RET(recvBuf == nullptr, HCCL_ERROR("[AllGatherV] recvBuf is null when single rank"), HCCL_E_PTR);
2001 : }
2002 1 : return HcclAllGatherV2(sendBuf, recvBuf, sendCount, dataType, comm, stream);
2003 : }
2004 18 : CHK_RET_AND_PRINT_IDE(HcomCheckOpParamV2(tag.c_str(), sendCount, dataType, stream), tag.c_str());
2005 10 : CHK_RET_AND_PRINT_IDE(HcomCheckVOpParamV2(rankId, rankSize, sendCount, recvCounts), tag.c_str());
2006 3 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
2007 :
2008 : /* 接口交互信息日志 */
2009 : char stackLogBufferV2[LOG_TMPBUF_SIZE];
2010 3 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2011 0 : s32 streamId = HrtGetStreamId(stream);
2012 0 : s32 deviceLogicId = HrtGetDevice();
2013 0 : u32 localRank = INVALID_VALUE_RANKID;
2014 0 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
2015 :
2016 0 : s32 ret = snprintf_s(stackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
2017 : "tag[%s], sendBuf[%p], recvBuf[%p], sendCount[%llu], recvCounts[%p], recvDispls[%p], "
2018 : "dataType[%s], localRank[%u], streamId[%d], deviceLogicId[%d]",
2019 : tag.c_str(), sendBuf, recvBuf, sendCount, recvCounts, recvDispls,
2020 0 : GetDataTypeEnumStrV2(dataType).c_str(), localRank, streamId, deviceLogicId);
2021 :
2022 0 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
2023 0 : std::string logInfo = "Entry-HcclAllGatherVV2:" + std::string(stackLogBufferV2);
2024 0 : if (isCapture) {
2025 0 : CHK_PTR_NULL(rtModel);
2026 : // 获取不到modelId会报错
2027 0 : CHK_RET(GetModelId(rtModel, modelId));
2028 0 : logInfo += ", model id[" + to_string(modelId) + "].";
2029 : }
2030 0 : communicator->GetTrace().Save(logInfo);
2031 0 : }
2032 :
2033 3 : u64* counts = static_cast<u64 *>(recvCounts);
2034 3 : u64 output = 0;
2035 9 : for(size_t index = 0; index < rankSize; index++){
2036 6 : output += counts[index];
2037 : }
2038 3 : if(output == 0){
2039 3 : HCCL_INFO("[%s] output[%llu] is equal to zero", __func__, output);
2040 1 : return HCCL_SUCCESS;
2041 : }
2042 2 : RPT_INPUT_ERR(recvBuf == nullptr, "EI0003", std::vector<std::string>({"ccl_op", "value", "parameter", "expect"}),\
2043 : std::vector<std::string>({"HcclAllGatherVV2", "nullptr", "recvBuf", "not nullptr"}));
2044 2 : CHK_PTR_NULL(recvBuf);
2045 : // opParams组装
2046 2 : Hccl::CollOpParams opParams;
2047 2 : opParams.opType = Hccl::OpType::ALLGATHERV;
2048 2 : opParams.dataType = HcclDataTypeToDataType(dataType);
2049 2 : opParams.dstRank = rankId;
2050 2 : opParams.sendBuf = sendBuf;
2051 2 : opParams.recvBuf = recvBuf;
2052 2 : opParams.count = sendCount;
2053 2 : opParams.vDataDes.counts = recvCounts;
2054 2 : opParams.vDataDes.displs = recvDispls;
2055 2 : opParams.vDataDes.dataType = HcclDataTypeToDataType(dataType);
2056 :
2057 2 : CHK_RET_AND_PRINT_IDE(communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream)), tag.c_str());
2058 :
2059 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2060 0 : HcclUs endut = TIME_NOW();
2061 : /* 关键状态记录 */
2062 0 : std::string endInfo = "HcclAllGatherVV2:success,take time: " +
2063 0 : std::to_string(DURATION_US(endut - startut).count()) + " us, tag: " + tag + std::string(stackLogBufferV2);
2064 0 : communicator->GetTrace().Save(endInfo);
2065 0 : }
2066 :
2067 2 : return HCCL_SUCCESS;
2068 9 : }
2069 :
2070 4 : HcclResult ValidateRank(uint32_t rank, Hccl::HcclCommunicator *communicator)
2071 : {
2072 4 : u32 rankSize{};
2073 4 : CHK_RET(communicator->GetRankSize(&rankSize));
2074 4 : u32 rankId{INVALID_VALUE_RANKID};
2075 4 : CHK_RET(communicator->GetRankId(rankId));
2076 4 : CHK_RET(HcomCheckUserRankV2(rankSize, rank));
2077 4 : CHK_PRT_RET(rankId == rank, HCCL_ERROR("same rank[%u] is not allowed", rank), HCCL_E_PARA);
2078 4 : return HCCL_SUCCESS;
2079 : }
2080 :
2081 2 : HcclResult HcclSendV2(
2082 : void *sendBuf, uint64_t count, HcclDataType dataType, uint32_t destRank, HcclComm comm, aclrtStream stream)
2083 : {
2084 2 : HcclUs startut = TIME_NOW();
2085 : bool isCapture;
2086 2 : rtModel_t rtModel = nullptr;
2087 2 : u32 modelId = 0;
2088 :
2089 2 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
2090 2 : const std::string tag = "SendRecv_" + communicator->GetId();
2091 :
2092 2 : CHK_RET(HcomCheckDataTypeV2(dataType));
2093 2 : CHK_RET(ValidateRank(destRank, communicator));
2094 2 : CHK_RET_AND_PRINT_IDE(HcomCheckOpParamV2(tag.c_str(), count, dataType, stream), tag.c_str());
2095 2 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
2096 :
2097 : /* 接口交互信息日志 */
2098 : char hcclSendStackLogBufferV2[LOG_TMPBUF_SIZE];
2099 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2100 1 : s32 streamId = HrtGetStreamId(stream);
2101 1 : s32 deviceLogicId = HrtGetDevice();
2102 1 : u32 localRank = INVALID_VALUE_RANKID;
2103 1 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
2104 :
2105 1 : s32 ret = snprintf_s(hcclSendStackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
2106 : "tag[%s], sendBuf[%p], count[%llu], dataType[%s], localRank[%u], streamId[%d], deviceLogicId[%d]",
2107 2 : tag.c_str(), sendBuf, count, GetDataTypeEnumStrV2(dataType).c_str(), localRank, streamId, deviceLogicId);
2108 :
2109 1 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
2110 1 : std::string logInfo = "Entry-HcclSendV2:" + std::string(hcclSendStackLogBufferV2);
2111 1 : if (isCapture) {
2112 0 : CHK_PTR_NULL(rtModel);
2113 : // 获取不到modelId会报错
2114 0 : CHK_RET(GetModelId(rtModel, modelId));
2115 0 : logInfo += ", model id[" + to_string(modelId) + "].";
2116 : }
2117 1 : communicator->GetTrace().Save(logInfo);
2118 1 : }
2119 :
2120 2 : static thread_local Hccl::CollOpParams opParams{};
2121 2 : opParams.opType = Hccl::OpType::SEND;
2122 2 : opParams.dataType = HcclDataTypeToDataType(dataType);
2123 2 : opParams.sendBuf = sendBuf;
2124 2 : opParams.recvBuf = nullptr;
2125 2 : opParams.count = count;
2126 2 : opParams.dstRank = destRank;
2127 2 : opParams.opTag = tag;
2128 :
2129 2 : CHK_RET_AND_PRINT_IDE(communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream)), tag.c_str());
2130 :
2131 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2132 1 : HcclUs endut = TIME_NOW();
2133 : /* 关键状态记录 */
2134 1 : std::string endInfo = "HcclSendV2:success,take time: " +
2135 2 : std::to_string(DURATION_US(endut - startut).count()) + " us, tag: " + tag + std::string(hcclSendStackLogBufferV2);
2136 1 : communicator->GetTrace().Save(endInfo);
2137 1 : }
2138 :
2139 2 : return HCCL_SUCCESS;
2140 2 : }
2141 :
2142 2 : HcclResult HcclRecvV2(
2143 : void *recvBuf, uint64_t count, HcclDataType dataType, uint32_t srcRank, HcclComm comm, aclrtStream stream)
2144 : {
2145 2 : HcclUs startut = TIME_NOW();
2146 : bool isCapture;
2147 2 : rtModel_t rtModel = nullptr;
2148 2 : u32 modelId = 0;
2149 :
2150 2 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
2151 2 : const std::string tag = "SendRecv_" + communicator->GetId();
2152 :
2153 2 : CHK_RET(HcomCheckDataTypeV2(dataType));
2154 2 : CHK_RET(ValidateRank(srcRank, communicator));
2155 2 : CHK_RET_AND_PRINT_IDE(HcomCheckOpParamV2(tag.c_str(), count, dataType, stream), tag.c_str());
2156 2 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
2157 :
2158 : /* 接口交互信息日志 */
2159 : char hcclRecvStackLogBufferV2[LOG_TMPBUF_SIZE];
2160 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2161 1 : s32 streamId = HrtGetStreamId(stream);
2162 1 : s32 deviceLogicId = HrtGetDevice();
2163 1 : u32 localRank = INVALID_VALUE_RANKID;
2164 1 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
2165 :
2166 1 : s32 ret = snprintf_s(hcclRecvStackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
2167 : "tag[%s], recvBuf[%p], count[%llu], dataType[%s], localRank[%u], streamId[%d], deviceLogicId[%d]",
2168 2 : tag.c_str(), recvBuf, count, GetDataTypeEnumStrV2(dataType).c_str(), localRank, streamId, deviceLogicId);
2169 :
2170 1 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
2171 1 : std::string logInfo = "Entry-HcclRecvV2:" + std::string(hcclRecvStackLogBufferV2);
2172 1 : if (isCapture) {
2173 0 : CHK_PTR_NULL(rtModel);
2174 : // 获取不到modelId会报错
2175 0 : CHK_RET(GetModelId(rtModel, modelId));
2176 0 : logInfo += ", model id[" + to_string(modelId) + "].";
2177 : }
2178 1 : communicator->GetTrace().Save(logInfo);
2179 1 : }
2180 :
2181 2 : static thread_local Hccl::CollOpParams opParams{};
2182 2 : opParams.opType = Hccl::OpType::RECV;
2183 2 : opParams.dataType = HcclDataTypeToDataType(dataType);
2184 2 : opParams.sendBuf = nullptr;
2185 2 : opParams.recvBuf = recvBuf;
2186 2 : opParams.count = count;
2187 2 : opParams.dstRank = srcRank;
2188 2 : opParams.opTag = tag;
2189 :
2190 2 : CHK_RET_AND_PRINT_IDE(communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream)), tag.c_str());
2191 :
2192 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2193 1 : HcclUs endut = TIME_NOW();
2194 : /* 关键状态记录 */
2195 1 : std::string endInfo = "HcclRecvV2:success,take time: " +
2196 2 : std::to_string(DURATION_US(endut - startut).count()) + " us, tag: " + tag + std::string(hcclRecvStackLogBufferV2);
2197 1 : communicator->GetTrace().Save(endInfo);
2198 1 : }
2199 :
2200 2 : return HCCL_SUCCESS;
2201 2 : }
2202 :
2203 9 : HcclResult HcclReduceScatterV2(void *sendBuf, void *recvBuf, uint64_t recvCount, HcclDataType dataType, HcclReduceOp op,
2204 : HcclComm comm, aclrtStream stream)
2205 : {
2206 9 : HcclUs startut = TIME_NOW();
2207 : bool isCapture;
2208 9 : rtModel_t rtModel = nullptr;
2209 9 : u32 modelId = 0;
2210 :
2211 9 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
2212 9 : const std::string tag = "ReduceScatter_" + communicator->GetId();
2213 :
2214 9 : CHK_RET_AND_PRINT_IDE(HcomCheckOpParamV2(tag.c_str(), recvCount, dataType, stream), tag.c_str());
2215 9 : CHK_RET_AND_PRINT_IDE(HcomCheckReductionOpV2(op), tag.c_str());
2216 21 : CHK_RET_AND_PRINT_IDE(HcomCheckReduceDataTypeV2(dataType, op), tag.c_str());
2217 7 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
2218 :
2219 : /* 接口交互信息日志 */
2220 : char stackLogBufferV2[LOG_TMPBUF_SIZE];
2221 7 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2222 3 : s32 streamId = HrtGetStreamId(stream);
2223 3 : s32 deviceLogicId = HrtGetDevice();
2224 3 : u32 localRank = INVALID_VALUE_RANKID;
2225 3 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
2226 :
2227 6 : s32 ret = snprintf_s(stackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
2228 : "tag[%s], sendBuf[%p], recvBuf[%p], recvCount[%llu], dataType[%s], op[%s],"
2229 : "localRank[%u], streamId[%d], deviceLogicId[%d]",
2230 9 : tag.c_str(), sendBuf, recvBuf, recvCount, GetDataTypeEnumStrV2(dataType).c_str(), GetReduceOpEnumStrV2(op).c_str(),
2231 : localRank, streamId, deviceLogicId);
2232 :
2233 3 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
2234 3 : std::string logInfo = "Entry-HcclReduceScatterV2:" + std::string(stackLogBufferV2);
2235 3 : if (isCapture) {
2236 0 : CHK_PTR_NULL(rtModel);
2237 : // 获取不到modelId会报错
2238 0 : CHK_RET(GetModelId(rtModel, modelId));
2239 0 : logInfo += ", model id[" + to_string(modelId) + "].";
2240 : }
2241 3 : communicator->GetTrace().Save(logInfo);
2242 3 : }
2243 :
2244 7 : static thread_local Hccl::CollOpParams opParams;
2245 7 : opParams.opType = Hccl::OpType::REDUCESCATTER;
2246 7 : opParams.dataType = HcclDataTypeToDataType(dataType);
2247 7 : opParams.reduceOp = HCCL_OP_REDUCE_MAP[op];
2248 7 : opParams.sendBuf = sendBuf;
2249 7 : opParams.recvBuf = recvBuf;
2250 7 : opParams.count = recvCount;
2251 7 : opParams.opTag = tag;
2252 :
2253 7 : CHK_RET_AND_PRINT_IDE(communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream)), tag.c_str());
2254 :
2255 7 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2256 3 : HcclUs endut = TIME_NOW();
2257 : /* 关键状态记录 */
2258 3 : std::string endInfo = "HcclReduceScatterV2:success,take time: " +
2259 6 : std::to_string(DURATION_US(endut - startut).count()) + " us, tag: " + tag + std::string(stackLogBufferV2);
2260 3 : communicator->GetTrace().Save(endInfo);
2261 3 : }
2262 :
2263 7 : return HCCL_SUCCESS;
2264 9 : }
2265 :
2266 10 : HcclResult HcclReduceScatterVV2(void *sendBuf, void *sendCounts, void *sendDispls, void *recvBuf, uint64_t recvCount,
2267 : HcclDataType dataType, HcclReduceOp op, HcclComm comm, aclrtStream stream)
2268 : {
2269 10 : HcclUs startut = TIME_NOW();
2270 : bool isCapture;
2271 10 : rtModel_t rtModel = nullptr;
2272 10 : u32 modelId = 0;
2273 :
2274 : // 获取通信域
2275 10 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
2276 10 : const std::string tag = "ReduceScatterV_" + communicator->GetId();
2277 :
2278 : // 获取rank信息
2279 : uint32_t rankId;
2280 10 : CHK_RET(communicator->GetRankId(rankId));
2281 : uint32_t rankSize;
2282 10 : CHK_RET(communicator->GetRankSize(&rankSize));
2283 : // 参数合法性校验
2284 10 : if (rankSize == 1) {
2285 : // rankSize为1时，退化为ReduceScatter
2286 : // 检查异常回退RSV的情况
2287 3 : if (recvCount == 0) {
2288 3 : HCCL_WARNING("[ReduceScatterV] recvCount is 0 when single rank");
2289 1 : return HCCL_SUCCESS;
2290 : } else {
2291 5 : CHK_PRT_RET(sendBuf == nullptr, HCCL_ERROR("[ReduceScatterV] sendBuf is null when single rank"), HCCL_E_PTR);
2292 1 : CHK_PRT_RET(recvBuf == nullptr, HCCL_ERROR("[ReduceScatterV] recvBuf is null when single rank"), HCCL_E_PTR);
2293 : }
2294 1 : return HcclReduceScatterV2(sendBuf, recvBuf, recvCount, dataType, op, comm, stream);
2295 : }
2296 13 : CHK_RET_AND_PRINT_IDE(HcomCheckOpParamV2(tag.c_str(), recvCount, dataType, stream), tag.c_str());
2297 6 : CHK_RET_AND_PRINT_IDE(HcomCheckReductionOpV2(op), tag.c_str());
2298 12 : CHK_RET_AND_PRINT_IDE(HcomCheckReduceDataTypeV2(dataType, op), tag.c_str());
2299 17 : CHK_RET_AND_PRINT_IDE(HcomCheckVOpParamV2(rankId, rankSize, recvCount, sendCounts), tag.c_str());
2300 3 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
2301 :
2302 : /* 接口交互信息日志 */
2303 : char stackLogBufferV2[LOG_TMPBUF_SIZE];
2304 3 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2305 0 : s32 streamId = HrtGetStreamId(stream);
2306 0 : s32 deviceLogicId = HrtGetDevice();
2307 0 : u32 localRank = INVALID_VALUE_RANKID;
2308 0 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
2309 :
2310 0 : s32 ret = snprintf_s(stackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
2311 : "tag[%s], sendBuf[%p], recvBuf[%p], sendCounts[%p], sendDispls[%p], recvCount[%llu], dataType[%s], op[%s],"
2312 : "localRank[%u], streamId[%d], deviceLogicId[%d]",
2313 : tag.c_str(), sendBuf, recvBuf, sendCounts, sendDispls, recvCount,
2314 0 : GetDataTypeEnumStrV2(dataType).c_str(), GetReduceOpEnumStrV2(op).c_str(), localRank, streamId, deviceLogicId);
2315 :
2316 0 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
2317 0 : std::string logInfo = "Entry-HcclReduceScatterVV2:" + std::string(stackLogBufferV2);
2318 0 : if (isCapture) {
2319 0 : CHK_PTR_NULL(rtModel);
2320 : // 获取不到modelId会报错
2321 0 : CHK_RET(GetModelId(rtModel, modelId));
2322 0 : logInfo += ", model id[" + to_string(modelId) + "].";
2323 : }
2324 0 : communicator->GetTrace().Save(logInfo);
2325 0 : }
2326 :
2327 3 : u64* counts = static_cast<u64 *>(sendCounts);
2328 3 : u64 inputCount = 0;
2329 9 : for(size_t index = 0; index < rankSize; index++){
2330 6 : inputCount += counts[index];
2331 : }
2332 3 : if(inputCount == 0){
2333 3 : HCCL_INFO("[%s] inputCount[%llu] is equal to zero", __func__, inputCount);
2334 1 : return HCCL_SUCCESS;
2335 : }
2336 2 : RPT_INPUT_ERR(sendBuf == nullptr, "EI0003", std::vector<std::string>({"ccl_op", "value", "parameter", "expect"}),\
2337 : std::vector<std::string>({"HcclReduceScatterVV2", "nullptr", "sendBuf", "not nullptr"}));
2338 2 : CHK_PTR_NULL(sendBuf);
2339 2 : if (op == HCCL_REDUCE_PROD) {
2340 0 : HCCL_ERROR("[Check][ReductionOp] Op:[HCCL_REDUCE_PROD] not supported, tag[%s]", tag.c_str());
2341 0 : return HCCL_E_NOT_SUPPORT;
2342 : }
2343 : // opParams组装
2344 2 : Hccl::CollOpParams opParams;
2345 2 : opParams.opType = Hccl::OpType::REDUCESCATTERV;
2346 2 : opParams.dataType = HcclDataTypeToDataType(dataType);
2347 2 : opParams.reduceOp = HcclReduceOpToReduceOp(op);
2348 2 : opParams.dstRank = rankId;
2349 2 : opParams.sendBuf = sendBuf;
2350 2 : opParams.recvBuf = recvBuf;
2351 2 : opParams.count = recvCount;
2352 2 : opParams.vDataDes.counts = sendCounts;
2353 2 : opParams.vDataDes.displs = sendDispls;
2354 2 : opParams.vDataDes.dataType = HcclDataTypeToDataType(dataType);
2355 :
2356 2 : CHK_RET_AND_PRINT_IDE(communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream)), tag.c_str());
2357 :
2358 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2359 0 : HcclUs endut = TIME_NOW();
2360 : /* 关键状态记录 */
2361 0 : std::string endInfo = "HcclReduceScatterVV2:success,take time: " +
2362 0 : std::to_string(DURATION_US(endut - startut).count()) + " us, tag: " + tag + std::string(stackLogBufferV2);
2363 0 : communicator->GetTrace().Save(endInfo);
2364 0 : }
2365 :
2366 2 : return HCCL_SUCCESS;
2367 10 : }
2368 :
2369 2 : HcclResult HcclBatchSendRecvV2(HcclSendRecvItem *sendRecvInfo, uint32_t itemNum, HcclComm comm, aclrtStream stream)
2370 : {
2371 2 : HcclUs startut = TIME_NOW();
2372 : bool isCapture;
2373 2 : rtModel_t rtModel = nullptr;
2374 2 : u32 modelId = 0;
2375 :
2376 2 : CHK_PTR_NULL(comm);
2377 2 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
2378 2 : const std::string tag = "HcclBatchSendRecvV2_" + communicator->GetId();
2379 2 : CHK_PTR_NULL(stream);
2380 2 : CHK_PTR_NULL(sendRecvInfo);
2381 2 : CHK_PRT_RET(itemNum == 0, HCCL_WARNING("[BatchSendRecv] taskList itemNum is zero."), HCCL_SUCCESS);
2382 2 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
2383 2 : u32 rankSize = 0;
2384 2 : CHK_RET(communicator->GetRankSize(&rankSize)); // 获取rankSize, 后续需要校验userrank是否在0到rankSize-1之间
2385 22 : for (uint32_t i = 0; i < itemNum; i++) {
2386 20 : if ((sendRecvInfo + i)->buf == nullptr) {
2387 2 : continue; // 支持数据量为0的场景，buf为空的跳过
2388 : }
2389 18 : CHK_RET_AND_PRINT_IDE(HcomCheckOpParamV2(tag.c_str(), (sendRecvInfo + i)->count, (sendRecvInfo + i)->dataType, stream), tag.c_str());
2390 18 : CHK_RET(HcomCheckUserRankV2(rankSize, (sendRecvInfo + i)->remoteRank));
2391 : }
2392 :
2393 : /* 记录接口交互信息日志 */
2394 : char stackLogBufferV2[LOG_TMPBUF_SIZE];
2395 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2396 1 : s32 streamId = HrtGetStreamId(stream);
2397 1 : s32 deviceLogicId = HrtGetDevice();
2398 1 : u32 localRank = INVALID_VALUE_RANKID;
2399 1 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
2400 :
2401 1 : s32 ret = snprintf_s(stackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
2402 : "tag[%s], itemNum[%u], localRank[%u], streamId[%d], deviceLogicId[%d]", tag.c_str(), itemNum, localRank, streamId, deviceLogicId);
2403 :
2404 1 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
2405 1 : std::string logInfo = "Entry-HcclBatchSendRecvV2:" + std::string(stackLogBufferV2);
2406 1 : if (isCapture) {
2407 0 : CHK_PTR_NULL(rtModel);
2408 0 : CHK_RET(GetModelId(rtModel, modelId));
2409 0 : logInfo += ", model id[" + to_string(modelId) + "].";
2410 : }
2411 1 : communicator->GetTrace().Save(logInfo);
2412 1 : }
2413 :
2414 2 : static thread_local Hccl::CollOpParams opParams;
2415 2 : opParams.opType = Hccl::OpType::BATCHSENDRECV;
2416 2 : opParams.batchSendRecvDataDes.sendRecvItemsPtr = static_cast<void *>(sendRecvInfo);
2417 2 : opParams.batchSendRecvDataDes.itemNum = itemNum;
2418 2 : opParams.dataType = HcclDataTypeToDataType(sendRecvInfo->dataType);
2419 :
2420 2 : CHK_RET_AND_PRINT_IDE(communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream)), tag.c_str());
2421 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2422 1 : HcclUs endut = TIME_NOW();
2423 1 : std::string endInfo = "HcclBatchSendRecvV2:success,take time: " +
2424 2 : std::to_string(DURATION_US(endut - startut).count()) + " us, tag: " + tag + std::string(stackLogBufferV2);
2425 1 : communicator->GetTrace().Save(endInfo);
2426 1 : }
2427 2 : return HCCL_SUCCESS;
2428 2 : }
2429 :
2430 : // 功能说明：推动式建链，超时退出
2431 2 : HcclResult WaitAllCommReady(s32 deviceLogicId)
2432 : {
2433 : try {
2434 2 : HrtSetDevice(deviceLogicId);
2435 2 : HcclCommInfoV2 &opbasedCommInfoV2 = GetCommInfoV2();
2436 : // 定义最大等待10秒
2437 2 : constexpr u32 waitTransportReadyTimeoutMs = 10 * 1000; // 待解决
2438 2 : auto timeout = std::chrono::milliseconds(waitTransportReadyTimeoutMs);
2439 2 : HcclUs startTime = std::chrono::steady_clock::now();
2440 :
2441 : // 创建锁，防止在建链过程中，依旧还在恢复通信域
2442 2 : std::unique_lock<std::mutex> groupParaLock(opbasedCommInfoV2.groupParamsLock);
2443 6 : HCCL_INFO("[%s] deviceLogicId[%d] start wait all comm ready", __func__, deviceLogicId);
2444 : // 轮巡调度，推动式建链
2445 : while (true) {
2446 2 : bool isAllCommReady = true;
2447 : Hccl::HcclCommunicator *communicator;
2448 : // 枚举所有通信域进行推动式建链
2449 5 : for (auto iter = opbasedCommInfoV2.hcclGroupMap.begin(); iter != opbasedCommInfoV2.hcclGroupMap.end(); iter++) {
2450 3 : CHK_PTR_NULL(iter->second.pComm);
2451 3 : communicator = static_cast<Hccl::HcclCommunicator *>(iter->second.pComm.get());
2452 3 : if (!communicator->IsCommReady()) {
2453 : // 只要任意Comm一个没有ready，整体建链结果为 false
2454 1 : isAllCommReady = false;
2455 : }
2456 : }
2457 2 : if (isAllCommReady) {
2458 1 : break;
2459 : }
2460 :
2461 : // 超时判断
2462 1 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
2463 3 : HCCL_ERROR("[%s]WaitAllCommReady timeout, deviceLogicId[%d].", __func__, deviceLogicId);
2464 1 : return HCCL_E_TIMEOUT;
2465 : }
2466 0 : }
2467 1 : HrtResetDevice(deviceLogicId);
2468 2 : } catch (HcclException &e) {
2469 0 : HCCL_ERROR(e.what());
2470 0 : return e.GetErrorCode();
2471 0 : } catch (std::exception &e) {
2472 0 : HCCL_ERROR(e.what());
2473 0 : return HCCL_E_INTERNAL;
2474 0 : } catch (...) {
2475 0 : HCCL_ERROR("Unknown error occurs!");
2476 0 : return HCCL_E_INTERNAL;
2477 0 : }
2478 :
2479 3 : HCCL_INFO("[%s] all comm ready.", __func__);
2480 1 : return HCCL_SUCCESS;
2481 : }
2482 :
2483 : #ifdef __cplusplus
2484 : extern "C" {
2485 : #endif // __cplusplus
2486 :
2487 0 : HcclResult HcclGetCclBuffer(HcclComm comm, uintptr_t &cclBufferAddr, size_t &cclBufferSize, HcclMemType &cclBufferMemType)
2488 : {
2489 0 : Hccl::HcclCommunicator* communicatorV2 = (static_cast<Hccl::HcclCommunicator*>(comm));
2490 0 : CHK_PTR_NULL(communicatorV2);
2491 0 : CHK_RET(communicatorV2->HcclGetCclBuffer(cclBufferAddr, cclBufferSize, cclBufferMemType));
2492 0 : return HCCL_SUCCESS;
2493 : }
2494 :
2495 1 : HcclResult HcclGetRawCommHandle(const char *commName, HcclComm *commHandle)
2496 : {
2497 1 : CHK_PTR_NULL(commName);
2498 1 : CHK_PTR_NULL(commHandle);
2499 :
2500 1 : HcclCommInfoV2 &opbasedCommInfoV2 = GetCommInfoV2();
2501 1 : std::unique_lock<std::mutex> lock(opbasedCommInfoV2.groupParamsLock);
2502 3 : HCCL_INFO("[HcclGetRawCommHandle] group:[%s]",commName);
2503 2 : auto iter = opbasedCommInfoV2.hcclGroupMap.find(commName);
2504 1 : if (iter == opbasedCommInfoV2.hcclGroupMap.end()) {
2505 0 : HCCL_ERROR("[HcclGetRawCommHandle] commName [%s] not found, please check.", commName);
2506 0 : return HCCL_E_PARA;
2507 : }
2508 2 : *commHandle = static_cast<HcclComm>(opbasedCommInfoV2.hcclGroupMap[commName].pComm.get());
2509 1 : return HCCL_SUCCESS;
2510 1 : }
2511 :
2512 2 : HcclResult HcclGetCcuTaskInfoLegacy(HcclComm comm, void *tilingData, void *ccuTaskGroup)
2513 : {
2514 2 : CHK_PTR_NULL(comm);
2515 2 : CHK_PTR_NULL(tilingData);
2516 2 : CHK_PTR_NULL(ccuTaskGroup);
2517 :
2518 2 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
2519 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2520 0 : HCCL_RUN_INFO("Entry-HcclGetCcuTaskInfo V950, commId[%s], tilingData[%p], ccuTaskGroup[%p]",
2521 : communicator->GetId().c_str(), tilingData, ccuTaskGroup);
2522 : }
2523 2 : auto ret = communicator->GetCcuTaskInfo(tilingData, ccuTaskGroup);
2524 2 : if (ret != HCCL_SUCCESS) {
2525 3 : HCCL_ERROR("HcclGetCcuTaskInfo ret[%d] commId[%s]", ret, communicator->GetId().c_str());
2526 1 : return HCCL_E_INTERNAL;
2527 : }
2528 :
2529 1 : return HCCL_SUCCESS;
2530 : }
2531 :
2532 2 : HcclResult HcclSnapshotSave(void *snapshotBuf, uint32_t size, uint32_t step)
2533 : {
2534 : // 快照保存只支持ranktable场景，不支持topo探测场景
2535 6 : HCCL_INFO("[%s] snapshot save start, size[%u], step[%u]", __func__, size, step);
2536 2 : CHK_PTR_NULL(snapshotBuf);
2537 : // 获取公共信息
2538 2 : HcclCommInfoV2 &opbasedCommInfoV2 = GetCommInfoV2();
2539 : // 获取公共信息里的step信息
2540 2 : u64 savedStep = opbasedCommInfoV2.step;
2541 : // 如果当前step不等于用户传入的step，直接返回报错
2542 2 : if (savedStep != step) {
2543 0 : HCCL_ERROR("[%s] step is not match, savedStep[%u], userInputStep[%u]", __func__, savedStep, step);
2544 0 : return HCCL_E_PARA;
2545 : }
2546 : // 获取保存的快照
2547 2 : Hccl::BinaryStream &savedSnapshotBuf = Hccl::SnapShotParser::GetInstance().GetSnapShotBuf();
2548 2 : uint32_t dataLen = savedSnapshotBuf.GetSize();
2549 6 : HCCL_INFO("[%s] savedSnapshotBuf data len[%u]", __func__, dataLen);
2550 :
2551 : // 获取保存的size
2552 2 : if ((dataLen + sizeof(dataLen) + sizeof(uint32_t)) != size) {
2553 : // 如果当前size不等于用户传入的size，直接返回报错
2554 3 : HCCL_ERROR("[%s] size is not match, userInputSize[%u] dataLen[%u]", __func__, size, dataLen);
2555 1 : return HCCL_E_PARA;
2556 : }
2557 1 : std::vector<char> data;
2558 1 : savedSnapshotBuf.DumpWithRevert(data);
2559 3 : HCCL_INFO("[%s] dump data size [%u]", __func__, data.size());
2560 1 : if (data.empty()) {
2561 3 : HCCL_INFO("[%s] dump data empty", __func__);
2562 1 : return HCCL_E_INTERNAL;
2563 : }
2564 :
2565 : // 计算快照crc值
2566 0 : uint32_t crcValue{0};
2567 0 : CHK_RET(Hccl::SnapShotParser::GetInstance().CalcBufCrc32(savedSnapshotBuf, crcValue));
2568 :
2569 : // 将快照大小拷贝到用户传入的内存中
2570 0 : s32 sRet = memcpy_s(snapshotBuf, size, static_cast<void *>(&dataLen), sizeof(dataLen));
2571 0 : CHK_PRT_RET(sRet != EOK, HCCL_ERROR("[%s] memcpy dataLen failed, return[%d]", __func__, sRet), HCCL_E_MEMORY);
2572 : // 将快照crc值拷贝到用户传入的内存中
2573 0 : sRet = memcpy_s(static_cast<char *>(snapshotBuf) + sizeof(dataLen), size - sizeof(dataLen), static_cast<void *>(&crcValue), sizeof(crcValue));
2574 0 : CHK_PRT_RET(sRet != EOK, HCCL_ERROR("[%s] memcpy crcValue failed, return[%d]", __func__, sRet), HCCL_E_MEMORY);
2575 : // 将快照拷贝到用户传入的内存中
2576 0 : sRet = memcpy_s(static_cast<char *>(snapshotBuf) + sizeof(dataLen) + sizeof(crcValue), dataLen, &data[0], dataLen);
2577 0 : CHK_PRT_RET(sRet != EOK, HCCL_ERROR("[%s] memcpy data failed, return[%d]", __func__, sRet), HCCL_E_MEMORY);
2578 0 : return HCCL_SUCCESS;
2579 1 : }
2580 :
2581 1 : void RecoverSnapshotCcuStatus(const std::shared_ptr<Hccl::SnapShotBuf>& savedSnapshotBuf)
2582 : {
2583 1 : HcclCommInfoV2 &opbasedCommInfoV2 = GetCommInfoV2();
2584 : // 通过进程锁看护，避免多个通信域同时占用CCU_MS
2585 1 : std::unique_lock<std::mutex> lock(opbasedCommInfoV2.groupParamsLock);
2586 :
2587 1 : for (auto useMsCommId : savedSnapshotBuf->ccuStatusSnapshot.useMsCommIds) {
2588 0 : opbasedCommInfoV2.ccuStatus.useMsCommIds.push_back(useMsCommId.data());
2589 : }
2590 :
2591 1 : for (auto useSchedCommId : savedSnapshotBuf->ccuStatusSnapshot.useSchedCommIds) {
2592 0 : opbasedCommInfoV2.ccuStatus.useSchedCommIds.push_back(useSchedCommId.data());
2593 : }
2594 : // 结果打印
2595 3 : HCCL_INFO("RecoverSnapshotCcuStatus ccuMsCommIds size[%lu] ccuSchedCommIds size[%lu]",
2596 : opbasedCommInfoV2.ccuStatus.useMsCommIds.size(), opbasedCommInfoV2.ccuStatus.useSchedCommIds.size());
2597 1 : for (auto useMsCommId : opbasedCommInfoV2.ccuStatus.useMsCommIds) {
2598 0 : HCCL_DEBUG("RecoverSnapshotCcuStatus useMsCommId[%s]", useMsCommId.c_str());
2599 0 : }
2600 1 : for (auto useSchedCommId : opbasedCommInfoV2.ccuStatus.useSchedCommIds) {
2601 0 : HCCL_DEBUG("RecoverSnapshotCcuStatus useSchedCommId[%s]", useSchedCommId.c_str());
2602 0 : }
2603 1 : }
2604 :
2605 1 : HcclResult HcclSnapshotRecoverAllComms(const char *clusterInfo, const char *changedInfo,
2606 : void *snapshotBuf, uint32_t snapshotBufSize)
2607 : {
2608 : (void)clusterInfo;
2609 : // 入参校验，clusterInfo和changedInfo暂时用不到，先不判空
2610 3 : HCCL_INFO("[%s] snapshot recover start.", __func__);
2611 1 : CHK_PTR_NULL(snapshotBuf);
2612 :
2613 : // 当前不支持在不下发算子的情况下重复调用recover接口
2614 1 : if (Hccl::SnapShotParser::GetInstance().GetIsNeedLoadOp()) {
2615 0 : HCCL_ERROR("[%s] snapshot recover failed, it is necessary to use it after load op, please check!", __func__);
2616 0 : return HCCL_E_INTERNAL;
2617 : }
2618 :
2619 1 : s32 deviceLogicId = HcclGetThreadDeviceId();
2620 :
2621 1 : CHK_RET(CallSingletons()); // 临时规避，在初始化通信域前声明单例保证时序
2622 :
2623 : // 获取公共信息
2624 1 : HcclCommInfoV2 &opbasedCommInfoV2 = GetCommInfoV2();
2625 : // 调用恢复快照公共信息函数，让字节流的struct数据恢复到savedSnapshotBuf
2626 1 : std::shared_ptr<Hccl::SnapShotBuf> savedSnapshotBuf = std::make_shared<Hccl::SnapShotBuf>();
2627 1 : CHK_RET(Hccl::SnapShotParser::GetInstance().ParseSnapshotToLocalBuff(snapshotBuf, snapshotBufSize, *savedSnapshotBuf));
2628 :
2629 1 : opbasedCommInfoV2.step = savedSnapshotBuf->snapShotPub.step;
2630 : // 将状态设置为RECOVERED
2631 1 : opbasedCommInfoV2.status = DeviceStatus::DEVICE_RECOVERED;
2632 :
2633 : // 创建通信域
2634 1 : opbasedCommInfoV2.pComm.reset(new (std::nothrow) Hccl::HcclCommunicator(
2635 1 : savedSnapshotBuf->snapshot.snapShotComm.commParams, &savedSnapshotBuf->snapshot.snapShotComm.config));
2636 1 : CHK_PTR_NULL(opbasedCommInfoV2.pComm);
2637 1 : opbasedCommInfoV2.pComm->RegisterAcceStateCallBack(CommunicatorCallback());
2638 : // 恢复全局通讯域
2639 3 : HCCL_INFO("[%s] recover global group.", __func__);
2640 1 : CHK_RET(opbasedCommInfoV2.pComm->RecoverComm(static_cast<void *>(&savedSnapshotBuf->snapshot.snapShotComm),
2641 : savedSnapshotBuf->snapShotPub.step, changedInfo));
2642 3 : HCCL_INFO("[%s] global group recover success.", __func__);
2643 1 : opbasedCommInfoV2.commParams = savedSnapshotBuf->snapshot.snapShotComm.commParams;
2644 :
2645 1 : HcclGroupParamsV2 params{};
2646 1 : params.pComm = opbasedCommInfoV2.pComm;
2647 2 : opbasedCommInfoV2.hcclGroupMap[savedSnapshotBuf->snapshot.groupName] = params;
2648 :
2649 : // 这里变成多线程后，封装成小函数
2650 2 : for (uint32_t index = 0; index < savedSnapshotBuf->groupNum; index++) {
2651 1 : std::string groupName(savedSnapshotBuf->subSnapshot[index].groupName);
2652 : // 别重复恢复全局通讯域
2653 1 : if (groupName == savedSnapshotBuf->snapshot.groupName) {
2654 0 : continue;
2655 : }
2656 1 : const Hccl::SnapShotSubComm *snapShotSubComm = &savedSnapshotBuf->subSnapshot[index].snapShotSubComm;
2657 : // 创建一个指针
2658 : std::shared_ptr<Hccl::HcclCommunicator> commImp = make_shared<Hccl::HcclCommunicator>(
2659 1 : snapShotSubComm->commParams, &snapShotSubComm->config);
2660 : // 循环创建恢复子通讯域
2661 3 : HCCL_INFO("[%s] recover sub group[%s].", __func__, groupName.c_str());
2662 1 : CHK_RET(opbasedCommInfoV2.pComm->RecoverSubComm(static_cast<const void *>(snapShotSubComm), commImp,
2663 : savedSnapshotBuf->snapShotPub.step));
2664 3 : HCCL_INFO("[%s] sub group[%s] recover success.", __func__, groupName.c_str());
2665 1 : HcclGroupParamsV2 params{};
2666 1 : params.pComm = commImp;
2667 : // 这里进行多线程后，加锁
2668 1 : opbasedCommInfoV2.hcclGroupMap[groupName] = params;
2669 1 : }
2670 1 : RecoverSnapshotCcuStatus(savedSnapshotBuf);
2671 3 : HCCL_INFO("[%s] snapshot recover success.", __func__);
2672 1 : std::unique_ptr<std::thread> waitReadyThread;
2673 1 : waitReadyThread.reset(new (std::nothrow) std::thread(&WaitAllCommReady, deviceLogicId));
2674 1 : if (waitReadyThread == nullptr) {
2675 0 : HCCL_ERROR("[%s] new waitReadyThread failed.", __func__);
2676 0 : return HCCL_E_INTERNAL;
2677 : }
2678 : // 异步方案目前未分析清楚，采用临时方案规避
2679 1 : waitReadyThread->join();
2680 1 : Hccl::SnapShotParser::GetInstance().SetIsNeedLoadOp(true);
2681 3 : HCCL_INFO("[%s] all comm ready.", __func__);
2682 1 : return HCCL_SUCCESS;
2683 1 : }
2684 3 : static HcclResult GetAllSnapShotStaticBuf(const std::shared_ptr<Hccl::HcclCommunicator> &pComm,
2685 : const std::map<std::string, std::shared_ptr<Hccl::HcclCommunicator>> &hcclGroupMap,
2686 : uint32_t step, Hccl::BinaryStream &buf)
2687 : {
2688 9 : HCCL_INFO("[%s] start", __func__);
2689 3 : if (!pComm->IsWorldGroup()) {
2690 0 : HCCL_ERROR("[%s] input comm is not hccl_world_group, please check!", __func__);
2691 0 : return HCCL_E_INTERNAL;
2692 : }
2693 : // 生成 静态流公共数据,直接往buf写
2694 3 : Hccl::SnapShotParser::GetInstance().SerializeCommVersionInfo(buf);
2695 : // 存储 全局通信域名称
2696 3 : buf << pComm->GetId();
2697 : // 存储 全局通信域 静态buf
2698 3 : Hccl::BinaryStream &pCommBuf = *(static_cast<Hccl::BinaryStream *>(pComm->GetStaticBinaryInfo()));
2699 3 : std::vector<char> pCommChars{};
2700 3 : pCommBuf.Dump(pCommChars);
2701 3 : for (auto c : pCommChars) {
2702 0 : buf << c;
2703 : }
2704 :
2705 3 : size_t pSubCommSize = hcclGroupMap.size() - 1;
2706 9 : HCCL_INFO("[%s] pSubCommSize[%u]", __func__, pSubCommSize);
2707 3 : buf << pSubCommSize; // 子通信域 静态buf 数量
2708 :
2709 3 : auto iter = hcclGroupMap.begin();
2710 6 : for (; iter != hcclGroupMap.end(); iter++) {
2711 3 : auto pSubCommName = iter->first;
2712 3 : if (pComm->GetId() == pSubCommName) {
2713 2 : continue;
2714 : }
2715 1 : buf << pSubCommName; // 子通信域 名字
2716 : // 获取子通信域 静态buf
2717 1 : auto groupComm = iter->second;
2718 1 : Hccl::BinaryStream &pSubCommBuf = *(static_cast<Hccl::BinaryStream *>(groupComm->GetStaticBinaryInfo()));
2719 1 : std::vector<char> pSubCommChars{};
2720 1 : pSubCommBuf.Dump(pSubCommChars);
2721 1 : for (auto c : pSubCommChars) {
2722 0 : buf << c;
2723 : }
2724 3 : }
2725 3 : buf << step;
2726 9 : HCCL_INFO("[%s] end", __func__);
2727 :
2728 3 : return HCCL_SUCCESS;
2729 3 : }
2730 :
2731 3 : static HcclResult GetAllSnapShotDynamicBuf(const std::shared_ptr<Hccl::HcclCommunicator> &pComm,
2732 : const std::map<std::string, std::shared_ptr<Hccl::HcclCommunicator>> &hcclGroupMap, Hccl::BinaryStream &buf)
2733 : {
2734 9 : HCCL_INFO("[%s] start", __func__);
2735 3 : CHK_PTR_NULL(pComm);
2736 : // 全局通信域 动态buf信息
2737 3 : CHK_RET(pComm->GetSnapShotDynamicBuf(static_cast<void *>(&buf)));
2738 :
2739 3 : size_t pSubCommSize = hcclGroupMap.size() - 1;
2740 3 : buf << pSubCommSize; // 子通信域 动态buf信息 数量
2741 9 : HCCL_INFO("[%s] hcclGroupMap size[%u]", __func__, hcclGroupMap.size());
2742 3 : auto iter = hcclGroupMap.begin();
2743 6 : for (; iter != hcclGroupMap.end(); iter++) {
2744 3 : auto pSubCommName = iter->first;
2745 3 : if (pComm->GetId() == pSubCommName) {
2746 2 : continue;
2747 : }
2748 : // 获取子通信域 建链邻居信息buf
2749 1 : CHK_RET(iter->second->GetSnapShotDynamicBuf(static_cast<void *>(&buf)));
2750 3 : }
2751 9 : HCCL_INFO("[%s] end", __func__);
2752 3 : return HCCL_SUCCESS;
2753 : }
2754 :
2755 3 : void GetSnapShotCcuStatusBuf(Hccl::BinaryStream &buf)
2756 : {
2757 3 : HcclCommInfoV2 &opbasedCommInfoV2 = GetCommInfoV2();
2758 : // 通过进程锁看护，避免多个通信域同时占用CCU_MS
2759 3 : std::unique_lock<std::mutex> lock(opbasedCommInfoV2.groupParamsLock);
2760 :
2761 3 : auto ccuStatus = opbasedCommInfoV2.ccuStatus;
2762 3 : buf << ccuStatus.useMsCommIds.size();
2763 9 : HCCL_INFO("useMsCommIds size is %u", ccuStatus.useMsCommIds.size());
2764 3 : for (auto useMsCommId : ccuStatus.useMsCommIds) {
2765 0 : buf << useMsCommId;
2766 0 : HCCL_INFO("useMsCommId is %s", useMsCommId.c_str());
2767 0 : }
2768 :
2769 3 : buf << ccuStatus.useSchedCommIds.size();
2770 9 : HCCL_INFO("useSchedCommIds size is %u", ccuStatus.useSchedCommIds.size());
2771 3 : for (auto useSchedCommId : ccuStatus.useSchedCommIds) {
2772 0 : buf << useSchedCommId;
2773 0 : HCCL_INFO("useSchedCommId is %s", useSchedCommId.c_str());
2774 0 : }
2775 3 : }
2776 :
2777 : // 获取快照占用buffer的大小 --给op_base API getSize调，内部生成完整长流。save后释放
2778 3 : HcclResult SnapshotGenerate(const std::shared_ptr<Hccl::HcclCommunicator> &pComm,
2779 : const std::map<std::string, std::shared_ptr<Hccl::HcclCommunicator>> &hcclGroupMap, uint32_t step, uint32_t *size)
2780 : {
2781 3 : CHK_PTR_NULL(size);
2782 3 : CHK_PTR_NULL(pComm);
2783 9 : HCCL_INFO("[%s] start", __func__);
2784 3 : Hccl::SnapShotParser::GetInstance().GetSnapShotBuf().Clear();
2785 3 : CHK_RET(GetAllSnapShotStaticBuf(pComm, hcclGroupMap, step, Hccl::SnapShotParser::GetInstance().GetSnapShotBuf()));
2786 3 : CHK_RET(GetAllSnapShotDynamicBuf(pComm, hcclGroupMap, Hccl::SnapShotParser::GetInstance().GetSnapShotBuf()));
2787 3 : GetSnapShotCcuStatusBuf(Hccl::SnapShotParser::GetInstance().GetSnapShotBuf());
2788 : // size = 流长度 + crc（u32）长度 + 存储流长度所需长度
2789 3 : uint32_t dataLen = static_cast<uint32_t>(Hccl::SnapShotParser::GetInstance().GetSnapShotBuf().GetSize());
2790 3 : *size = dataLen + sizeof(dataLen) + sizeof(uint32_t); // 快照头上保存一个总长度和crc长度
2791 9 : HCCL_INFO("[%s] end, size[%u]", __func__, *size);
2792 3 : return HCCL_SUCCESS;
2793 : }
2794 :
2795 2 : HcclResult HcclSnapshotGetBufSize(uint32_t step, uint32_t *size)
2796 : {
2797 : // 校验DevType
2798 6 : HCCL_INFO("[%s] start", __func__);
2799 2 : Hccl::DevType devType = HrtGetDeviceType();
2800 2 : if (devType != DevType::DEV_TYPE_950 && devType != DevType::DEV_TYPE_960) {
2801 0 : HCCL_INFO("[%s] Get buffer size not support in this device type[%d]", __func__, devType);
2802 0 : return HCCL_E_NOT_SUPPORT;
2803 : }
2804 :
2805 : // step保存在g_opbasedCommInfoV2，save的时候校验，防止GetBufSize和Save的step不一致
2806 2 : HcclCommInfoV2 &opbasedCommInfoV2 = GetCommInfoV2();
2807 2 : opbasedCommInfoV2.step = step;
2808 :
2809 2 : std::map<std::string, std::shared_ptr<Hccl::HcclCommunicator>> subCommMap;
2810 4 : for (auto hcclGroupMap : opbasedCommInfoV2.hcclGroupMap) {
2811 2 : subCommMap.insert(std::make_pair(hcclGroupMap.first, hcclGroupMap.second.pComm));
2812 2 : }
2813 2 : return SnapshotGenerate(opbasedCommInfoV2.pComm, subCommMap, step, size);
2814 2 : }
2815 :
2816 1 : HcclResult HcclGetTopoDescV2()
2817 : {
2818 3 : HCCL_ERROR("Current chip type does not support GetTopoDesc.");
2819 1 : return HCCL_E_NOT_SUPPORT;
2820 : }
2821 :
2822 1 : HcclResult HcclGetCommAsyncErrorV2()
2823 : {
2824 3 : HCCL_WARNING("HcclGetCommAsyncErrorV2 is not support!");
2825 1 : return HCCL_SUCCESS;
2826 : }
2827 :
2828 0 : HcclResult HcclSetConfigV2(HcclConfig config, HcclConfigValue configValue)
2829 : {
2830 : (void)(config);
2831 : (void)(configValue);
2832 0 : HCCL_WARNING("DETERMINISTIC_ENABLE is default option in 950! Can not set.");
2833 0 : return HCCL_SUCCESS;
2834 : }
2835 0 : HcclResult HcclGetConfigV2(HcclConfig config, HcclConfigValue *configValue)
2836 : {
2837 : (void)(config);
2838 0 : constexpr int32_t DETERMINISTIC_ENABLE = 1; // A5支持确定性，不需要配置
2839 0 : (*configValue).value = DETERMINISTIC_ENABLE;
2840 0 : HCCL_WARNING("DETERMINISTIC_ENABLE is default option in 950!");
2841 0 : return HCCL_SUCCESS;
2842 : }
2843 :
2844 1 : HcclResult HcclGetRankGraphV2(const HcclComm *comm, void **rankGraph)
2845 : {
2846 1 : Hccl::HcclCommunicator* communicatorV2 = (static_cast<Hccl::HcclCommunicator*>(*comm));
2847 1 : communicatorV2->GetRankGraphV2(*rankGraph);
2848 1 : return HCCL_SUCCESS;
2849 : }
2850 :
2851 0 : HcclResult HcommFlushV2()
2852 : {
2853 0 : HCCL_INFO("[HcommFlushV2]");
2854 0 : return FlushManager::GetInstance().Flush();
2855 : }
2856 :
2857 0 : HcclResult CommGetCCLBufSizeCfgV2(HcclComm comm, uint64_t *cclBufSize)
2858 : {
2859 0 : HCCL_RUN_INFO("Entry-CommGetCCLBufSizeCfg V950");
2860 0 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
2861 0 : CHK_RET(communicator->GetConfigInCCLbufferSize(cclBufSize));
2862 0 : return HCCL_SUCCESS;
2863 : }
2864 :
2865 0 : HcclResult HcclGetNetLayersV2(HcclComm comm, uint32_t **netLayers, uint32_t *netLayerNum)
2866 : {
2867 0 : HCCL_RUN_INFO("Entry-HcclGetNetLayersV2 V950");
2868 0 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
2869 0 : auto ret = communicator->GetNetLayers(netLayers, netLayerNum);
2870 0 : if (ret != HCCL_SUCCESS) {
2871 0 : HCCL_ERROR("HcclGetNetLayersV2 get netLayers from communicator failed, commId[%s], ret[%d]", communicator->GetId().c_str(), ret);
2872 0 : return HCCL_E_NOT_FOUND; // 查询失败返回
2873 : }
2874 0 : return HCCL_SUCCESS;
2875 : }
2876 :
2877 0 : HcclResult HcclGetInstSizeByNetLayerV2(HcclComm comm, uint32_t netLayer, uint32_t *rankNum)
2878 : {
2879 0 : HCCL_RUN_INFO("Entry-HcclGetInstSizeByNetLayerV2 V950");
2880 0 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
2881 0 : auto ret = communicator->GetInstSizeByNetLayer(netLayer, rankNum);
2882 0 : if (ret != HCCL_SUCCESS) {
2883 0 : HCCL_ERROR("HcclGetInstSizeByNetLayerV2 get InstSize from communicator failed at netLayer[%u], commId[%s], ret[%d]",
2884 : netLayer, communicator->GetId().c_str(), ret);
2885 0 : return HCCL_E_NOT_FOUND; // 查询失败返回
2886 : }
2887 : /* 关键状态记录 */
2888 0 : HCCL_INFO("HcclGetInstSizeByNetLayerV2 success, netLayer[%u], rankNum[%u], commId[%s]", netLayer, *rankNum, communicator->GetId().c_str());
2889 0 : return HCCL_SUCCESS;
2890 : }
2891 :
2892 2 : HcclResult HcclGetInstRanksByNetLayerV2(HcclComm comm, uint32_t netLayer, uint32_t **ranks, uint32_t *rankNum)
2893 : {
2894 6 : HCCL_RUN_INFO("Entry-HcclGetInstRanksByNetLayerV2 V950");
2895 2 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
2896 2 : auto ret = communicator->GetInstRanksByNetLayer(netLayer, ranks, rankNum);
2897 2 : if (ret != HCCL_SUCCESS) {
2898 3 : HCCL_ERROR("HcclGetInstRanksByNetLayerV2 get ranks from communicator failed at netLayer[%u], commId[%s], ret[%d]",
2899 : netLayer, communicator->GetId().c_str(), ret);
2900 1 : return HCCL_E_NOT_FOUND; // 查询失败返回
2901 : }
2902 : /* 关键状态记录 */
2903 3 : HCCL_INFO("HcclGetInstRanksByNetLayerV2 success, netLayer[%u], rankNum[%u], commId[%s]", netLayer, *rankNum, communicator->GetId().c_str());
2904 1 : return HCCL_SUCCESS;
2905 : }
2906 :
2907 2 : HcclResult HcclGetInstTopoTypeByNetLayerV2(HcclComm comm, uint32_t netLayer, uint32_t *topoType)
2908 : {
2909 6 : HCCL_RUN_INFO("Entry-HcclGetInstTopoTypeByNetLayer V950");
2910 2 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
2911 2 : auto ret = communicator->GetInstTopoTypeByNetLayer(netLayer, topoType);
2912 2 : if (ret != HCCL_SUCCESS) {
2913 3 : HCCL_ERROR("HcclGetInstTopoTypeByNetLayerV2 get topoType from communicator failed at netLayer[%u], commId[%s], ret[%d]",
2914 : netLayer, communicator->GetId().c_str(), ret);
2915 1 : return HCCL_E_NOT_FOUND;
2916 : }
2917 : /* 关键状态记录 */
2918 3 : HCCL_INFO("HcclGetInstTopoTypeByNetLayer success, netLayer[%u] topoType[%u] commId[%s]", netLayer, *topoType, communicator->GetId().c_str());
2919 1 : return HCCL_SUCCESS;
2920 : }
2921 :
2922 2 : HcclResult HcclGetInstSizeListByNetLayerV2(HcclComm comm, uint32_t netLayer, uint32_t **instSizeList,
2923 : uint32_t *listSize)
2924 : {
2925 6 : HCCL_RUN_INFO("Entry-HcclGetInstSizeListByNetLayer V950");
2926 2 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
2927 2 : auto ret = communicator->GetInstSizeListByNetLayer(netLayer, instSizeList, listSize);
2928 2 : if (ret != HCCL_SUCCESS) {
2929 3 : HCCL_ERROR("HcclGetInstSizeListByNetLayerV2 get netInstance size from communicator failed at netLayer[%u], commId[%s], ret[%d]",
2930 : netLayer, communicator->GetId().c_str(), ret);
2931 1 : return HCCL_E_NOT_FOUND;
2932 : }
2933 : /* 关键状态记录 */
2934 3 : HCCL_INFO("HcclGetInstSizeListByNetLayer success, netLayer[%u] listSize[%u] commId[%s]", netLayer, *listSize, communicator->GetId().c_str());
2935 1 : return HCCL_SUCCESS;
2936 : }
2937 :
2938 2 : HcclResult HcclGetLinksV2(HcclComm comm, uint32_t netLayer, uint32_t srcRank, uint32_t dstRank, CommLink **linkList,
2939 : uint32_t *listSize)
2940 : {
2941 6 : HCCL_RUN_INFO("Entry-HcclGetInstSizeListByNetLayer V950");
2942 2 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
2943 2 : auto ret = communicator->GetLinks(netLayer, srcRank, dstRank, linkList, listSize);
2944 2 : if (ret != HCCL_SUCCESS) {
2945 3 : HCCL_ERROR("HcclGetLinksV2 get links from communicator failed at netLayer[%u], commId[%s], ret[%d]",
2946 : netLayer, communicator->GetId().c_str(), ret);
2947 1 : return HCCL_E_NOT_FOUND;
2948 : }
2949 : /* 关键状态记录 */
2950 3 : HCCL_INFO("HcclGetLinks success, netLayer[%u], srcRank[%u], dstRank[%u], listSize[%u], commId[%s]", netLayer,
2951 : srcRank, dstRank, *listSize, communicator->GetId().c_str());
2952 1 : return HCCL_SUCCESS;
2953 : }
2954 :
2955 2 : HcclResult HcclGetTopoInstsByLayerV2(HcclComm comm, uint32_t netLayer, uint32_t **topoInsts, uint32_t *topoInstNum)
2956 : {
2957 6 : HCCL_RUN_INFO("Entry-HcclGetTopoInstsByLayer V950");
2958 2 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
2959 2 : auto ret = communicator->GetTopoInstsByLayer(netLayer, topoInsts, topoInstNum);
2960 2 : if (ret != HCCL_SUCCESS) {
2961 3 : HCCL_ERROR("HcclGetTopoInstsByLayer get topoInsts from communicator failed at netLayer[%u], commId[%s], ret[%d]",
2962 : netLayer, communicator->GetId().c_str(), ret);
2963 1 : return HCCL_E_NOT_FOUND;
2964 : }
2965 : /* 关键状态记录 */
2966 3 : HCCL_INFO("HcclGetTopoInstsByLayer success, netLayer[%u], topoInstNum[%u], commId[%s]", netLayer, *topoInstNum, communicator->GetId().c_str());
2967 1 : return HCCL_SUCCESS;
2968 : }
2969 :
2970 2 : HcclResult HcclGetTopoTypeV2(HcclComm comm, uint32_t netLayer, uint32_t topoInstId, CommTopo *topoType)
2971 : {
2972 6 : HCCL_RUN_INFO("Entry-HcclGetInstSizeListByNetLayer V950");
2973 2 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
2974 2 : auto ret = communicator->GetTopoType(netLayer, topoInstId, topoType);
2975 2 : if (ret != HCCL_SUCCESS) {
2976 3 : HCCL_ERROR("HcclGetTopoType get topoType from communicator failed at netLayer[%u], commId[%s], ret[%d]",
2977 : netLayer, communicator->GetId().c_str(), ret);
2978 1 : return HCCL_E_NOT_FOUND;
2979 : }
2980 : /* 关键状态记录 */
2981 3 : HCCL_INFO("HcclGetInstSizeListByNetLayer success, netLayer[%u] topoType[%u] commId[%s]", netLayer, *topoType, communicator->GetId().c_str());
2982 1 : return HCCL_SUCCESS;
2983 : }
2984 :
2985 2 : HcclResult HcclGetRanksByTopoInstV2(HcclComm comm, uint32_t netLayer, uint32_t topoInstId, uint32_t **ranks,
2986 : uint32_t *rankNum)
2987 : {
2988 6 : HCCL_RUN_INFO("Entry-HcclGetRanksByTopoInst V950");
2989 2 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
2990 2 : auto ret = communicator->GetRanksByTopoInst(netLayer, topoInstId, ranks, rankNum);
2991 2 : if (ret != HCCL_SUCCESS) {
2992 3 : HCCL_ERROR("HcclGetTopoInstsByLayer get ranks from communicator failed at netLayer[%u], commId[%s], ret[%d]",
2993 : netLayer, communicator->GetId().c_str(), ret);
2994 1 : return HCCL_E_NOT_FOUND;
2995 : }
2996 : /* 关键状态记录 */
2997 3 : HCCL_INFO("HcclGetRanksByTopoInst success, netLayer[%u] rankNum[%u]", netLayer, *rankNum);
2998 1 : return HCCL_SUCCESS;
2999 : }
3000 :
3001 2 : HcclResult HcclRankGraphGetEndpointNumV2(HcclComm comm, uint32_t layer, uint32_t topoInstId, uint32_t *num)
3002 : {
3003 6 : HCCL_RUN_INFO("Entry-HcclRankGraphGetEndpointNum V950");
3004 2 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
3005 2 : auto ret = communicator->GetEndpointNum(layer, topoInstId, num);
3006 2 : if (ret != HCCL_SUCCESS) {
3007 3 : HCCL_ERROR("HcclRankGraphGetEndpointNum get endpoint num from communicator failed at netLayer[%u] with topoInstId[%u], commId[%s], ret[%d]",
3008 : layer, topoInstId, communicator->GetId().c_str(), ret);
3009 1 : return HCCL_E_NOT_FOUND;
3010 : }
3011 1 : return HCCL_SUCCESS;
3012 : }
3013 :
3014 3 : HcclResult HcclRankGraphGetEndpointDescV2(HcclComm comm, uint32_t layer, uint32_t topoInstId, uint32_t *descNum, EndpointDesc *endpointDesc)
3015 : {
3016 9 : HCCL_RUN_INFO("Entry-HcclRankGraphGetEndpointDesc V950");
3017 3 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
3018 3 : auto ret = communicator->GetEndpointDesc(layer, topoInstId, descNum, endpointDesc);
3019 3 : if (ret != HCCL_SUCCESS) {
3020 3 : HCCL_ERROR("HcclRankGraphGetEndpointDesc get endpoint desc from communicator failed at netLayer[%u], commId[%s], ret[%d]",
3021 : layer, communicator->GetId().c_str(), ret);
3022 1 : return HCCL_E_NOT_FOUND;
3023 : }
3024 2 : return HCCL_SUCCESS;
3025 : }
3026 :
3027 1 : HcclResult HcclRankGraphGetEndpointInfoV2(HcclComm comm, uint32_t rankId, const EndpointDesc *endpointDesc, EndpointAttr endpointAttr, uint32_t infoLen, void *info)
3028 : {
3029 3 : HCCL_RUN_INFO("Entry-HcclRankGraphGetEndpointInfo V950");
3030 1 : Hccl::HcclCommunicator *communicator = static_cast<Hccl::HcclCommunicator *>(comm);
3031 1 : auto ret = communicator->GetEndpointInfo(rankId, endpointDesc, endpointAttr, infoLen, info);
3032 1 : if (ret != HCCL_SUCCESS) {
3033 0 : HCCL_ERROR("HcclRankGraphGetEndpointInfo get info from communicator failed with endpointAttr [%d], commId[%s], ret[%d]",
3034 : static_cast<s32>(endpointAttr), communicator->GetId().c_str(), ret);
3035 0 : return HCCL_E_NOT_FOUND;
3036 : }
3037 1 : return HCCL_SUCCESS;
3038 : }
3039 :
3040 1 : HcclResult HcclCommWorkingDevNicSetV2(const HcclComm comm, uint32_t *ranks, bool *useBackup, uint32_t nRanks)
3041 : {
3042 : (void) comm;
3043 : (void) ranks;
3044 : (void) useBackup;
3045 : (void) nRanks;
3046 3 : HCCL_ERROR("HcclCommWorkingDevNicSetV2 not support V950.");
3047 1 : return HCCL_E_NOT_SUPPORT;
3048 : }
3049 :
3050 1 : HcclResult HcclCommSetMemoryRangeV2(const HcclComm comm, void *baseVirPtr, size_t size, size_t alignment, uint64_t flags)
3051 : {
3052 : (void) comm;
3053 : (void) baseVirPtr;
3054 : (void) size;
3055 : (void) alignment;
3056 : (void) flags;
3057 3 : HCCL_ERROR("HcclCommSetMemoryRangeV2 not support V950.");
3058 1 : return HCCL_E_NOT_SUPPORT;
3059 : }
3060 :
3061 1 : HcclResult HcclCommUnsetMemoryRangeV2(const HcclComm comm, void *baseVirPtr)
3062 : {
3063 : (void) comm;
3064 : (void) baseVirPtr;
3065 3 : HCCL_ERROR("HcclCommUnsetMemoryRangeV2 not support V950.");
3066 1 : return HCCL_E_NOT_SUPPORT;
3067 : }
3068 :
3069 1 : HcclResult HcclCommActivateCommMemoryV2(const HcclComm comm, void *virPtr, size_t size, size_t offset, void* handle, uint64_t flags)
3070 : {
3071 : (void) comm;
3072 : (void) virPtr;
3073 : (void) size;
3074 : (void) offset;
3075 : (void) handle;
3076 : (void) flags;
3077 3 : HCCL_ERROR("HcclCommActivateCommMemoryV2 not support V950.");
3078 1 : return HCCL_E_NOT_SUPPORT;
3079 : }
3080 :
3081 1 : HcclResult HcclCommDeactivateCommMemoryV2(const HcclComm comm, void *virPtr)
3082 : {
3083 : (void) comm;
3084 : (void) virPtr;
3085 3 : HCCL_ERROR("HcclCommDeactivateCommMemoryV2 not support V950.");
3086 1 : return HCCL_E_NOT_SUPPORT;
3087 : }
3088 :
3089 0 : uint32_t HcclGetCommConfigCapabilityV2()
3090 : {
3091 0 : return static_cast<uint32_t>(HCCL_COMM_CONFIG_RETRY);
3092 : }
3093 :
3094 : #ifdef __cplusplus
3095 : }
3096 : #endif // __cplusplus
|
