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