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