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 DpuSteamId");
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 : HCCL_ERROR(
1048 : "[Init][%s]HcclCreateSubCommConfigV2 failed, deviceLogicId[%d], devPhyId[%d], sub comm[%s], world comm[%s]"
1049 : "return[0x%016llx]",
1050 : __func__, logicDevId, devPhyId, subCommIdStr.c_str(), commId.c_str(), HCCL_ERROR_CODE(ret));
1051 0 : (void)HcclCommDestroyV2(subCommunicator.get());
1052 0 : *subComm = nullptr;
1053 0 : return ret;
1054 : }
1055 : /* 关键状态记录 */
1056 6 : HCCL_RUN_INFO(
1057 : "[Create][Group]create group[%s] success, deviceLogicId[%d], devPhyId[%d], take time [%lld]us",
1058 : subCommIdStr.c_str(), logicDevId, devPhyId, DURATION_US(TIME_NOW() - startut));
1059 2 : return HCCL_SUCCESS;
1060 3 : }
1061 :
1062 1 : HcclResult HcclGetRankIdV2(HcclComm comm, uint32_t* rank)
1063 : {
1064 3 : HCCL_RUN_INFO("Entry-HcclGetRankId V950");
1065 1 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
1066 1 : auto ret = communicator->GetRankId(*rank);
1067 1 : if (ret != HcclResult::HCCL_SUCCESS) {
1068 0 : return HCCL_E_INTERNAL;
1069 : }
1070 : /* 关键状态记录 */
1071 3 : HCCL_RUN_INFO(
1072 : "Entry-HcclGetRankId V950 success, comm[%s], rankIdPtr[%p], rankId[%u]", communicator->GetId().c_str(), rank,
1073 : *rank);
1074 1 : return HCCL_SUCCESS;
1075 : }
1076 :
1077 1 : HcclResult HcclGetCommNameV2(HcclComm commHandle, char* commName)
1078 : {
1079 1 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(commHandle);
1080 1 : if (communicator == nullptr) {
1081 0 : HCCL_ERROR("HcclGetCommNameV2 communicator is nullptr");
1082 0 : return HCCL_E_PTR;
1083 : }
1084 3 : HCCL_INFO(
1085 : "HcclGetCommNameV2 commId[%s], commId size[%zu], input commName ptr=[%p]", communicator->GetId().c_str(),
1086 : communicator->GetId().size(), commName);
1087 1 : s32 ret = strncpy_s(
1088 1 : commName, ROOTINFO_INDENTIFIER_MAX_LENGTH, communicator->GetId().c_str(), communicator->GetId().size() + 1);
1089 1 : CHK_PRT_RET(
1090 : ret != EOK,
1091 : HCCL_ERROR("HcclGetCommName str copy fail. return[%d], commId[%s]", ret, communicator->GetId().c_str()),
1092 : HCCL_E_INTERNAL);
1093 1 : return HCCL_SUCCESS;
1094 : }
1095 :
1096 9 : HcclResult HcclGetRankSizeV2(HcclComm comm, uint32_t* rankSize)
1097 : {
1098 9 : CHK_PTR_NULL(comm);
1099 9 : CHK_PTR_NULL(rankSize);
1100 9 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
1101 9 : s32 deviceLogicId = HcclGetThreadDeviceId();
1102 9 : s32 devPhyId = HrtGetDevicePhyIdByIndex(deviceLogicId);
1103 27 : HCCL_RUN_INFO(
1104 : "Entry-HcclGetRankSize V950, commId[%s], deviceLogicId[%d], devPhyId[%d]", communicator->GetId().c_str(),
1105 : deviceLogicId, devPhyId);
1106 9 : auto ret = communicator->GetRankSize(rankSize);
1107 9 : if (ret != HCCL_SUCCESS) {
1108 0 : HCCL_ERROR("HcclGetRankSizeV2 failed, rankSize[%u], commId[%s]", *rankSize, communicator->GetId().c_str());
1109 0 : return HCCL_E_INTERNAL;
1110 : }
1111 : /* 关键状态记录 */
1112 27 : HCCL_RUN_INFO(
1113 : "Entry-HcclGetRankSize V950 success, comm[%s], rankSizePtr[%p], rankSize[%u]", communicator->GetId().c_str(),
1114 : rankSize, *rankSize);
1115 9 : return HCCL_SUCCESS;
1116 : }
1117 :
1118 2 : HcclResult HcclAlltoAllVCV2(
1119 : const void* sendBuf, const void* sendCountMatrix, HcclDataType sendType, const void* recvBuf, HcclDataType recvType,
1120 : HcclComm comm, rtStream_t stream)
1121 : {
1122 2 : HcclUs startut = TIME_NOW();
1123 : bool isCapture;
1124 2 : rtModel_t rtModel = nullptr;
1125 2 : u32 modelId = 0;
1126 2 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
1127 :
1128 2 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
1129 2 : const std::string tag = "HCCL_ALLTOALLVC_" + communicator->GetId();
1130 :
1131 2 : CHK_RET_AND_PRINT_IDE(HcomCheckOpParamV2(tag.c_str(), 0, sendType, stream), tag.c_str());
1132 2 : CHK_RET_AND_PRINT_IDE(HcomCheckDataTypeV2(recvType), tag.c_str());
1133 2 : u32 rankSize = 0;
1134 2 : CHK_RET(communicator->GetRankSize(&rankSize));
1135 2 : u32 myRank = INVALID_VALUE_RANKID;
1136 2 : CHK_RET(communicator->GetRankId(myRank));
1137 2 : bool isEmpty = false;
1138 2 : CHK_RET(HcomCheckAlltoAllVCEmptyV2(sendBuf, sendCountMatrix, recvBuf, rankSize, isEmpty));
1139 2 : if (isEmpty) {
1140 6 : HCCL_INFO("[HcclAlltoAllVCV2] sendCountMatrix is Empty");
1141 2 : return HCCL_SUCCESS;
1142 : }
1143 0 : CHK_RET(HcomCheckAlltoAllVCExternalMemV2(sendBuf, sendCountMatrix, recvBuf, rankSize, myRank));
1144 :
1145 : u64 sendCountMatrixHash;
1146 0 : HcomGetHashFromSendCountMatrixV2(sendCountMatrixHash, sendCountMatrix, rankSize, tag);
1147 : /* 接口交互信息日志 */
1148 : char stackLogBufferV2[LOG_TMPBUF_SIZE];
1149 0 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1150 0 : s32 streamId = HrtGetStreamId(stream);
1151 0 : s32 deviceLogicId = HrtGetDevice();
1152 0 : u32 localRank = INVALID_VALUE_RANKID;
1153 0 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
1154 :
1155 0 : s32 ret = snprintf_s(
1156 : stackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
1157 : "tag[%s], sendBuf[%p], sendCountMatrixHash[%llu], sendType[%s], recvBuf[%p],"
1158 : "recvType[%s], localRank[%u], streamId[%d], deviceLogicId[%d]",
1159 0 : tag.c_str(), sendBuf, sendCountMatrixHash, GetDataTypeEnumStrV2(sendType).c_str(), recvBuf,
1160 0 : GetDataTypeEnumStrV2(recvType).c_str(), localRank, streamId, deviceLogicId);
1161 :
1162 0 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
1163 0 : std::string logInfo = "Entry-HcclAlltoAllVCV2:" + std::string(stackLogBufferV2);
1164 0 : if (isCapture) {
1165 0 : CHK_PTR_NULL(rtModel);
1166 : // 获取不到modelId会报错
1167 0 : CHK_RET(GetModelId(rtModel, modelId));
1168 0 : logInfo += ", model id[" + to_string(modelId) + "].";
1169 : }
1170 0 : communicator->GetTrace().Save(logInfo);
1171 0 : }
1172 :
1173 0 : static thread_local Hccl::CollOpParams opParams;
1174 0 : opParams.opType = Hccl::OpType::ALLTOALLVC;
1175 0 : opParams.sendBuf = const_cast<void*>(sendBuf);
1176 0 : opParams.recvBuf = const_cast<void*>(recvBuf);
1177 0 : opParams.all2AllVCDataDes.sendCountMatrix = const_cast<void*>(sendCountMatrix);
1178 0 : opParams.all2AllVCDataDes.sendType = HcclDataTypeToDataType(sendType);
1179 0 : opParams.all2AllVCDataDes.recvType = HcclDataTypeToDataType(recvType);
1180 0 : opParams.dataType = HcclDataTypeToDataType(sendType);
1181 0 : CHK_RET_AND_PRINT_IDE(communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream)), tag.c_str());
1182 :
1183 0 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1184 0 : HcclUs endut = TIME_NOW();
1185 : /* 关键状态记录 */
1186 : std::string endInfo
1187 0 : = "HcclAlltoAllVCV2:success,take time: " + std::to_string(DURATION_US(endut - startut).count()) + " us,"
1188 0 : + std::string(stackLogBufferV2);
1189 0 : communicator->GetTrace().Save(endInfo);
1190 0 : }
1191 :
1192 0 : return HCCL_SUCCESS;
1193 2 : }
1194 :
1195 5 : HcclResult HcclReduceV2(
1196 : void* sendBuf, void* recvBuf, uint64_t count, HcclDataType dataType, HcclReduceOp op, uint32_t root, HcclComm comm,
1197 : aclrtStream stream)
1198 : {
1199 5 : HcclUs startut = TIME_NOW();
1200 : bool isCapture;
1201 5 : rtModel_t rtModel = nullptr;
1202 5 : u32 modelId = 0;
1203 :
1204 5 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
1205 5 : const std::string tag = "Reduce_" + communicator->GetId();
1206 :
1207 5 : CHK_RET_AND_PRINT_IDE(HcomCheckOpParamV2(tag.c_str(), count, dataType, stream), tag.c_str());
1208 5 : CHK_RET_AND_PRINT_IDE(HcomCheckReductionOpV2(op), tag.c_str());
1209 11 : CHK_RET_AND_PRINT_IDE(HcomCheckReduceDataTypeV2(dataType, op), tag.c_str());
1210 4 : u32 rankSize = INVALID_VALUE_RANKSIZE;
1211 4 : CHK_RET_AND_PRINT_IDE(communicator->GetRankSize(&rankSize), tag.c_str());
1212 4 : CHK_RET_AND_PRINT_IDE(HcomCheckUserRankV2(rankSize, root), tag.c_str());
1213 4 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
1214 :
1215 : /* 接口交互信息日志 */
1216 : char stackLogBufferV2[LOG_TMPBUF_SIZE];
1217 4 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1218 1 : s32 streamId = HrtGetStreamId(stream);
1219 1 : s32 deviceLogicId = HrtGetDevice();
1220 1 : u32 localRank = INVALID_VALUE_RANKID;
1221 1 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
1222 :
1223 2 : s32 ret = snprintf_s(
1224 : stackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
1225 : "tag[%s], sendBuf[%p], recvBuf[%p], count[%llu], dataType[%s], op[%s], root[%u],"
1226 : "localRank[%u], streamId[%d], deviceLogicId[%d]",
1227 2 : tag.c_str(), sendBuf, recvBuf, count, GetDataTypeEnumStrV2(dataType).c_str(),
1228 2 : GetReduceOpEnumStrV2(op).c_str(), root, localRank, streamId, deviceLogicId);
1229 :
1230 1 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
1231 1 : std::string logInfo = "Entry-HcclReduceV2:" + std::string(stackLogBufferV2);
1232 1 : if (isCapture) {
1233 0 : CHK_PTR_NULL(rtModel);
1234 : // 获取不到modelId会报错
1235 0 : CHK_RET(GetModelId(rtModel, modelId));
1236 0 : logInfo += ", model id[" + to_string(modelId) + "].";
1237 : }
1238 1 : communicator->GetTrace().Save(logInfo);
1239 1 : }
1240 :
1241 4 : static thread_local Hccl::CollOpParams opParams;
1242 4 : opParams.opType = Hccl::OpType::REDUCE;
1243 4 : opParams.dataType = HcclDataTypeToDataType(dataType);
1244 4 : opParams.reduceOp = HCCL_OP_REDUCE_MAP[op];
1245 4 : opParams.sendBuf = sendBuf;
1246 4 : opParams.recvBuf = recvBuf;
1247 4 : opParams.count = count;
1248 4 : opParams.root = root;
1249 4 : opParams.opTag = tag;
1250 4 : CHK_RET_AND_PRINT_IDE(communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream)), tag.c_str());
1251 :
1252 4 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1253 1 : HcclUs endut = TIME_NOW();
1254 : /* 关键状态记录 */
1255 2 : std::string endInfo = "HcclReduceV2:success,take time: " + std::to_string(DURATION_US(endut - startut).count())
1256 3 : + " us," + std::string(stackLogBufferV2);
1257 1 : communicator->GetTrace().Save(endInfo);
1258 1 : }
1259 :
1260 4 : return HCCL_SUCCESS;
1261 5 : }
1262 :
1263 8 : HcclResult HcclAllReduceV2(
1264 : void* sendBuf, void* recvBuf, uint64_t count, HcclDataType dataType, HcclReduceOp op, HcclComm comm,
1265 : aclrtStream stream)
1266 : {
1267 8 : HcclUs startut = TIME_NOW();
1268 : bool isCapture;
1269 8 : rtModel_t rtModel = nullptr;
1270 8 : u32 modelId = 0;
1271 :
1272 8 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
1273 8 : const std::string tag = "AllReduce_" + communicator->GetId();
1274 :
1275 8 : CHK_RET_AND_PRINT_IDE(HcomCheckOpParamV2(tag.c_str(), count, dataType, stream), tag.c_str());
1276 8 : CHK_RET_AND_PRINT_IDE(HcomCheckReductionOpV2(op), tag.c_str());
1277 14 : CHK_RET_AND_PRINT_IDE(HcomCheckReduceDataTypeV2(dataType, op), tag.c_str());
1278 7 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
1279 :
1280 : /* 接口交互信息日志 */
1281 : char stackLogBufferV2[LOG_TMPBUF_SIZE];
1282 7 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1283 1 : s32 streamId = HrtGetStreamId(stream);
1284 1 : s32 deviceLogicId = HrtGetDevice();
1285 1 : u32 localRank = INVALID_VALUE_RANKID;
1286 1 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
1287 :
1288 2 : s32 ret = snprintf_s(
1289 : stackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
1290 : "tag[%s], sendBuf[%p], recvBuf[%p], count[%llu], dataType[%s], op[%s], localRank[%u], streamId[%d],"
1291 : "comm[%p], deviceLogicId[%d]",
1292 2 : tag.c_str(), sendBuf, recvBuf, count, GetDataTypeEnumStrV2(dataType).c_str(),
1293 2 : GetReduceOpEnumStrV2(op).c_str(), localRank, streamId, comm, deviceLogicId);
1294 :
1295 1 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
1296 :
1297 1 : std::string logInfo = "Entry-HcclAllReduceV2: " + std::string(stackLogBufferV2);
1298 1 : if (isCapture) {
1299 0 : CHK_PTR_NULL(rtModel);
1300 : // 获取不到modelId会报错
1301 0 : CHK_RET(GetModelId(rtModel, modelId));
1302 0 : logInfo += ", model id[" + to_string(modelId) + "].";
1303 : }
1304 1 : communicator->GetTrace().Save(logInfo);
1305 1 : }
1306 :
1307 7 : static thread_local Hccl::CollOpParams opParams;
1308 7 : opParams.opType = Hccl::OpType::ALLREDUCE;
1309 7 : opParams.dataType = HcclDataTypeToDataType(dataType);
1310 7 : opParams.reduceOp = HCCL_OP_REDUCE_MAP[op];
1311 7 : opParams.sendBuf = sendBuf;
1312 7 : opParams.recvBuf = recvBuf;
1313 7 : opParams.count = count;
1314 7 : opParams.opTag = tag;
1315 7 : CHK_RET_AND_PRINT_IDE(communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream)), tag.c_str());
1316 :
1317 7 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1318 1 : HcclUs endut = TIME_NOW();
1319 : /* 关键状态记录 */
1320 : std::string endInfo
1321 2 : = "HcclAllReduceV2:success,take time: " + std::to_string(DURATION_US(endut - startut).count()) + " us,"
1322 3 : + std::string(stackLogBufferV2);
1323 1 : communicator->GetTrace().Save(endInfo);
1324 1 : }
1325 :
1326 7 : return HCCL_SUCCESS;
1327 8 : }
1328 :
1329 : HcclResult
1330 2 : HcclBroadcastV2(void* buf, uint64_t count, HcclDataType dataType, uint32_t root, HcclComm comm, aclrtStream stream)
1331 : {
1332 2 : HcclUs startut = TIME_NOW();
1333 : bool isCapture;
1334 2 : rtModel_t rtModel = nullptr;
1335 2 : u32 modelId = 0;
1336 :
1337 2 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
1338 2 : const std::string tag = "Broadcast_" + communicator->GetId();
1339 :
1340 2 : CHK_RET_AND_PRINT_IDE(HcomCheckOpParamV2(tag.c_str(), count, dataType, stream), tag.c_str());
1341 2 : u32 rankSize = INVALID_VALUE_RANKSIZE;
1342 2 : CHK_RET_AND_PRINT_IDE(communicator->GetRankSize(&rankSize), tag.c_str());
1343 2 : CHK_RET_AND_PRINT_IDE(HcomCheckUserRankV2(rankSize, root), tag.c_str());
1344 2 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
1345 :
1346 : /* 接口交互信息日志 */
1347 : char stackLogBufferV2[LOG_TMPBUF_SIZE];
1348 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1349 1 : s32 streamId = HrtGetStreamId(stream);
1350 1 : s32 deviceLogicId = HrtGetDevice();
1351 1 : u32 localRank = INVALID_VALUE_RANKID;
1352 1 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
1353 :
1354 1 : s32 ret = snprintf_s(
1355 : stackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
1356 : "tag[%s], buf[%p], count[%llu], dataType[%s], root[%u], localRank[%u], streamId[%d], deviceLogicId[%d]",
1357 2 : tag.c_str(), buf, count, GetDataTypeEnumStrV2(dataType).c_str(), root, localRank, streamId, deviceLogicId);
1358 :
1359 1 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
1360 1 : std::string logInfo = "Entry-HcclBroadcastV2:" + std::string(stackLogBufferV2);
1361 1 : if (isCapture) {
1362 0 : CHK_PTR_NULL(rtModel);
1363 : // 获取不到modelId会报错
1364 0 : CHK_RET(GetModelId(rtModel, modelId));
1365 0 : logInfo += ", model id[" + to_string(modelId) + "].";
1366 : }
1367 1 : communicator->GetTrace().Save(logInfo);
1368 1 : }
1369 :
1370 2 : static thread_local Hccl::CollOpParams opParams;
1371 2 : opParams.opType = Hccl::OpType::BROADCAST;
1372 2 : opParams.dataType = HcclDataTypeToDataType(dataType);
1373 2 : opParams.sendBuf = buf;
1374 2 : opParams.recvBuf = buf;
1375 2 : opParams.count = count;
1376 2 : opParams.root = root;
1377 2 : opParams.opTag = tag;
1378 2 : CHK_RET_AND_PRINT_IDE(communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream)), tag.c_str());
1379 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1380 1 : HcclUs endut = TIME_NOW();
1381 : /* 关键状态记录 */
1382 : std::string endInfo
1383 2 : = "HcclBroadcastV2:success,take time: " + std::to_string(DURATION_US(endut - startut).count()) + " us,"
1384 3 : + std::string(stackLogBufferV2);
1385 1 : communicator->GetTrace().Save(endInfo);
1386 1 : }
1387 :
1388 2 : return HCCL_SUCCESS;
1389 2 : }
1390 :
1391 2 : HcclResult HcclBarrierV2(HcclComm comm, aclrtStream stream)
1392 : {
1393 2 : HcclUs startut = TIME_NOW();
1394 6 : HCCL_INFO("HcclBarrierV2 V82");
1395 : // AllReduce入参定义
1396 2 : HcclDataType dataType = HCCL_DATA_TYPE_FP32;
1397 2 : HcclReduceOp op = HCCL_REDUCE_SUM;
1398 2 : const uint64_t count = 8;
1399 2 : void* sendBuf = nullptr;
1400 2 : void* recvBuf = nullptr;
1401 2 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
1402 2 : s32 deviceLogicId = HcclGetThreadDeviceId();
1403 2 : s32 devPhyId = HrtGetDevicePhyIdByIndex(deviceLogicId);
1404 6 : HCCL_RUN_INFO(
1405 : "Entry-HcclBarrier V950, commId[%s], deviceLogicId[%d], devPhyId[%d]", communicator->GetId().c_str(),
1406 : deviceLogicId, devPhyId);
1407 : // 申请Device内存
1408 2 : auto ret = communicator->CreateBarrierMemory(sendBuf, recvBuf, count);
1409 2 : if (ret != HCCL_SUCCESS) {
1410 1 : return ret;
1411 : }
1412 : // 同通信域同算子复用tag
1413 1 : const string tag = "AllReduce_" + communicator->GetId();
1414 :
1415 1 : CHK_RET_AND_PRINT_IDE(HcomCheckOpParamV2(tag.c_str(), count, dataType, stream), tag.c_str());
1416 :
1417 1 : CHK_RET_AND_PRINT_IDE(HcomCheckReductionOpV2(op), tag.c_str());
1418 :
1419 1 : CHK_RET_AND_PRINT_IDE(HcomCheckReduceDataTypeV2(dataType, op), tag.c_str());
1420 :
1421 1 : static thread_local Hccl::CollOpParams opParams;
1422 1 : opParams.opType = Hccl::OpType::ALLREDUCE;
1423 1 : opParams.dataType = HcclDataTypeToDataType(dataType);
1424 1 : opParams.reduceOp = Hccl::ReduceOp::SUM;
1425 :
1426 1 : opParams.sendBuf = sendBuf;
1427 1 : opParams.recvBuf = recvBuf;
1428 1 : opParams.count = count;
1429 1 : opParams.opTag = tag;
1430 1 : ret = communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream));
1431 3 : HCCL_RUN_INFO(
1432 : "Entry-HcclBarrier V950 success, take time [%lld]us, commId[%s], deviceLogicId[%d]",
1433 : DURATION_US(TIME_NOW() - startut), communicator->GetId().c_str(), deviceLogicId);
1434 1 : return ret;
1435 1 : }
1436 :
1437 1 : HcclResult HcclGetHeterogModeV2(HcclComm comm, HcclHeterogMode* mode)
1438 : {
1439 : (void)comm;
1440 1 : *mode = HCCL_HETEROG_MODE_HOMOGENEOUS;
1441 3 : HCCL_INFO("[HcclGetHeterogModeV2] 950 only support homogeneous chip mode");
1442 1 : return HCCL_SUCCESS;
1443 : }
1444 :
1445 1 : HcclResult HcclCommSuspendV2(HcclComm comm)
1446 : {
1447 1 : CHK_PTR_NULL(comm);
1448 3 : HCCL_ERROR("HcclCommSuspend V950 not support suspend");
1449 :
1450 1 : return HCCL_E_NOT_SUPPORT;
1451 : }
1452 :
1453 6 : HcclResult HcclAllocComResourceByTilingV2(HcclComm comm, const void* stream, void* mc2Tiling, void** commContext)
1454 : {
1455 9 : CHK_PTR_NULL(comm);
1456 8 : CHK_PTR_NULL(stream);
1457 7 : CHK_PTR_NULL(mc2Tiling);
1458 6 : CHK_PTR_NULL(commContext);
1459 2 : HcclUs startut = TIME_NOW();
1460 2 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
1461 :
1462 2 : uint32_t* pVersion = reinterpret_cast<uint32_t*>(mc2Tiling);
1463 2 : u32 localRank = INVALID_VALUE_RANKID;
1464 : char stackLogBufferV2[LOG_TMPBUF_SIZE];
1465 2 : string commIdentifier = communicator->GetId();
1466 2 : const std::string tag = "HcclAllocComResourceByTilingV2_" + communicator->GetId();
1467 2 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
1468 : /* 接口交互信息日志 */
1469 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1470 1 : s32 ret = snprintf_s(
1471 : stackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U, "commIdentifier[%s], version[%u]",
1472 : commIdentifier.c_str(), *pVersion);
1473 1 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, commIdentifier[%s].", commIdentifier.c_str()));
1474 :
1475 2 : std::string logInfo = "MC2 create resource by tiling: localRank[" + std::to_string(localRank) + "]"
1476 3 : + std::string(stackLogBufferV2);
1477 1 : communicator->GetTrace().Save(logInfo);
1478 1 : }
1479 :
1480 2 : HcclResult ret = communicator->AllocCommResource(mc2Tiling, commContext);
1481 2 : CHK_PRT_RET(
1482 : ret != HCCL_SUCCESS,
1483 : HCCL_ERROR(
1484 : "[HcclAllocComResourceByTilingV2]AllocCommResource fail, errNo[%d], commIdentifier[%s]", ret,
1485 : commIdentifier.c_str()),
1486 : ret);
1487 :
1488 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1489 1 : HcclUs endut = TIME_NOW();
1490 : /* 关键状态记录 */
1491 2 : std::string endInfo = "MC2 create resource take time [" + std::to_string(DURATION_US(endut - startut).count())
1492 3 : + "]us, localRank[" + std::to_string(localRank) + "] " + std::string(stackLogBufferV2);
1493 1 : communicator->GetTrace().Save(endInfo);
1494 1 : }
1495 :
1496 2 : return HCCL_SUCCESS;
1497 2 : }
1498 :
1499 2 : HcclResult HcclGetOpArgsV2(void** opArgs)
1500 : {
1501 2 : CHK_PTR_NULL(opArgs);
1502 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1503 0 : HCCL_RUN_INFO("Entry-HcclGetOpArgs V950, start malloc opArgs in %p", opArgs);
1504 : }
1505 2 : HcclOpArgs* opArgsMem = (HcclOpArgs*)malloc(sizeof(HcclOpArgs));
1506 2 : if (opArgsMem == nullptr) {
1507 0 : HCCL_ERROR("[HcclGetOpArgs] malloc HcclOpArgs mem fail, please check.");
1508 0 : return HCCL_E_INTERNAL;
1509 : }
1510 2 : opArgsMem->Init();
1511 2 : *opArgs = opArgsMem;
1512 6 : HCCL_RUN_INFO("HcclGetOpArgs malloc HcclOpArgs success, please fill mem[%p->%p] in it.", opArgs, *opArgs);
1513 2 : return HCCL_SUCCESS;
1514 : }
1515 :
1516 2 : HcclResult HcclFreeOpArgsV2(void* opArgs)
1517 : {
1518 2 : CHK_PTR_NULL(opArgs);
1519 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1520 0 : HCCL_RUN_INFO("Entry-HcclFreeOpArgs V950, free opArgs[%p]", opArgs);
1521 : }
1522 2 : free(opArgs);
1523 2 : opArgs = nullptr;
1524 2 : return HCCL_SUCCESS;
1525 : }
1526 :
1527 2 : HcclResult HcclSetOpSrcDataTypeV2(void* opArgs, uint8_t srcDataType)
1528 : {
1529 2 : CHK_PTR_NULL(opArgs);
1530 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1531 0 : HCCL_RUN_INFO("Entry-HcclSetOpSrcDataType V950, opArgs[%p] set srcDataType[%u]", opArgs, srcDataType);
1532 : }
1533 2 : HcclOpArgs* opArgsPtr = static_cast<HcclOpArgs*>(opArgs);
1534 2 : if (srcDataType >= (sizeof(MC2_DATA_TYPE) / sizeof(MC2_DATA_TYPE[0]))) {
1535 3 : HCCL_ERROR("HcclSetOpSrcDataType set srcDataType[%u] error, it's invalid.", srcDataType);
1536 1 : return HCCL_E_PARA;
1537 : }
1538 1 : opArgsPtr->srcDataType = MC2_DATA_TYPE[srcDataType];
1539 1 : return HCCL_SUCCESS;
1540 : }
1541 :
1542 2 : HcclResult HcclSetOpDstDataTypeV2(void* opArgs, uint8_t dstDataType)
1543 : {
1544 2 : CHK_PTR_NULL(opArgs);
1545 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1546 0 : HCCL_RUN_INFO("Entry-HcclSetOpDstDataType V950, opArgs[%p] set dstDataType[%u]", opArgs, dstDataType);
1547 : }
1548 2 : HcclOpArgs* opArgsPtr = static_cast<HcclOpArgs*>(opArgs);
1549 2 : if (dstDataType >= (sizeof(MC2_DATA_TYPE) / sizeof(MC2_DATA_TYPE[0]))) {
1550 3 : HCCL_ERROR("HcclSetOpDstDataType set dstDataType[%u] error, it's invalid.", dstDataType);
1551 1 : return HCCL_E_PARA;
1552 : }
1553 1 : opArgsPtr->dstDataType = MC2_DATA_TYPE[dstDataType];
1554 1 : return HCCL_SUCCESS;
1555 : }
1556 :
1557 2 : HcclResult HcclSetOpReduceTypeV2(void* opArgs, uint32_t reduceType)
1558 : {
1559 2 : CHK_PTR_NULL(opArgs);
1560 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1561 0 : HCCL_RUN_INFO("Entry-HcclSetOpReduceType V950, opArgs[%p] set reduceType[%u]", opArgs, reduceType);
1562 : }
1563 2 : HcclOpArgs* opArgsPtr = static_cast<HcclOpArgs*>(opArgs);
1564 2 : if (reduceType >= (sizeof(MC2_REDUCE_TYPE) / sizeof(MC2_REDUCE_TYPE[0]))) {
1565 3 : HCCL_ERROR("HcclSetOpReduceType set reduceType[%u] error, it's invalid.", reduceType);
1566 1 : return HCCL_E_PARA;
1567 : }
1568 1 : opArgsPtr->reduceType = MC2_REDUCE_TYPE[reduceType];
1569 1 : return HCCL_SUCCESS;
1570 : }
1571 :
1572 2 : HcclResult HcclSetOpCountV2(void* opArgs, uint64_t count)
1573 : {
1574 2 : CHK_PTR_NULL(opArgs);
1575 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1576 0 : HCCL_RUN_INFO("Entry-HcclSetOpCount V950, opArgs[%p] set count[%llu]", opArgs, count);
1577 : }
1578 2 : HcclOpArgs* opArgsPtr = static_cast<HcclOpArgs*>(opArgs);
1579 5 : CHK_RET(HcomCheckCountV2(count));
1580 1 : opArgsPtr->count = count;
1581 1 : return HCCL_SUCCESS;
1582 : }
1583 :
1584 2 : HcclResult HcclSetOpAlgConfigV2(void* opArgs, char* algConfig)
1585 : {
1586 2 : CHK_PTR_NULL(opArgs);
1587 2 : CHK_PTR_NULL(algConfig);
1588 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1589 0 : HCCL_RUN_INFO("Entry-HcclSetOpAlgConfig V950, opArgs[%p]", opArgs);
1590 : }
1591 2 : HcclOpArgs* opArgsPtr = static_cast<HcclOpArgs*>(opArgs);
1592 2 : s32 ret = strcpy_s(opArgsPtr->algConfig, ALG_CONFIG_SIZE, algConfig);
1593 2 : if (ret != EOK) {
1594 3 : HCCL_ERROR(
1595 : "[HcclSetOpAlgConfig]strcpy_s algConfig failed! result %d, the algConfig len must be less than %u", ret,
1596 : ALG_CONFIG_SIZE);
1597 1 : return HCCL_E_PARA;
1598 : }
1599 1 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1600 0 : HCCL_RUN_INFO(
1601 : "HcclSetOpAlgConfig set opArgs[%p] algConfig success, algConfig is [%s]", opArgs, opArgsPtr->algConfig);
1602 : }
1603 1 : return HCCL_SUCCESS;
1604 : };
1605 :
1606 1 : HcclResult HcclSetOpCommEngineV2(void* opArgs, uint8_t commEngine)
1607 : {
1608 1 : CHK_PTR_NULL(opArgs);
1609 1 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1610 0 : HCCL_RUN_INFO("Entry-HcclSetOpCommEngine V950, commEngine[%u]", commEngine);
1611 : }
1612 1 : HcclOpArgs* opArgsPtr = static_cast<HcclOpArgs*>(opArgs);
1613 1 : opArgsPtr->commEngine = HcclAccelerator(static_cast<HcclAccelerator::Value>(commEngine));
1614 1 : return HCCL_SUCCESS;
1615 : }
1616 :
1617 1 : HcclResult HcclCommResPrepareWithOpMode(
1618 : Hccl::HcclCommunicator* communicator, const std::string& opName, HcclOpArgs* opArgs, void** addr)
1619 : {
1620 1 : CHK_PTR_NULL(communicator);
1621 1 : CHK_PTR_NULL(opArgs);
1622 1 : CHK_PTR_NULL(addr);
1623 1 : OpType opType = GetOpTypeV2(opName);
1624 1 : if (opType == OpType::OPTYPEINVALID) {
1625 0 : HCCL_ERROR(
1626 : "[HcclCommResPrepareWithOpMode] The opName %s match opType is %s", opName.c_str(),
1627 : opType.Describe().c_str());
1628 0 : return HCCL_E_PARA;
1629 : }
1630 :
1631 1 : std::string opTag = opName + communicator->GetId() + "_mc2";
1632 1 : static thread_local Hccl::CollOpParams opParams;
1633 1 : opParams.opType = opType;
1634 1 : opParams.reduceOp = opArgs->reduceType;
1635 1 : opParams.dataType = opArgs->srcDataType;
1636 1 : opParams.outputDataType = opArgs->dstDataType;
1637 1 : opParams.count = opArgs->count;
1638 1 : opParams.opTag = opTag;
1639 1 : opParams.algConfig = std::string(opArgs->algConfig);
1640 1 : opParams.isMc2 = true;
1641 1 : opParams.commEngine = opArgs->commEngine;
1642 1 : return communicator->AllocCollOpResource(opParams, addr);
1643 1 : }
1644 :
1645 1 : HcclResult HcclCommResPrepareV2(HcclComm comm, char* opName, void* opArgs, void** addr)
1646 : {
1647 1 : CHK_PTR_NULL(comm);
1648 1 : CHK_PTR_NULL(opName);
1649 1 : CHK_PTR_NULL(opArgs);
1650 1 : CHK_PTR_NULL(addr);
1651 1 : std::string opNameStr(opName);
1652 1 : opNameStr = opNameStr.substr(0, MAX_OP_NAME_SIZE);
1653 1 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
1654 1 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1655 0 : HCCL_RUN_INFO(
1656 : "Entry-HcclCommResPrepare V950, opName[%s], opArgs addr[%p], commId[%s]", opNameStr.c_str(), opArgs,
1657 : communicator->GetId().c_str());
1658 : }
1659 1 : return HcclCommResPrepareWithOpMode(communicator, opNameStr, static_cast<HcclOpArgs*>(opArgs), addr);
1660 1 : }
1661 :
1662 2 : HcclResult HcclDevMemAcquireV2(HcclComm comm, const char* memTag, uint64_t* size, void** addr, bool* newCreated)
1663 : {
1664 2 : CHK_PTR_NULL(comm);
1665 2 : CHK_PTR_NULL(memTag);
1666 2 : CHK_PTR_NULL(size);
1667 2 : CHK_PTR_NULL(addr);
1668 2 : CHK_PTR_NULL(newCreated);
1669 2 : std::string memTagStr = "";
1670 2 : if (memTag != nullptr) {
1671 : char tmpMemTag[MAX_MEM_TAG_SIZE];
1672 2 : s32 ret = strcpy_s(tmpMemTag, MAX_MEM_TAG_SIZE, memTag);
1673 2 : if (ret != EOK) {
1674 0 : HCCL_ERROR(
1675 : "[HcclDevMemAcquire] strcpy_s memTag failed! result %d, the memTag len must be less than %u", ret,
1676 : MAX_MEM_TAG_SIZE);
1677 0 : return HCCL_E_PARA;
1678 : }
1679 4 : memTagStr = std::string(tmpMemTag);
1680 : }
1681 2 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
1682 2 : CHK_RET(communicator->GetDevMemWorkSpace(memTagStr, size, addr, newCreated));
1683 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1684 0 : HCCL_RUN_INFO(
1685 : "Entry-HcclDevMemAcquire V950, memTag[%s], addr[%p], size[%llu], commId[%s]", memTagStr.c_str(), *addr,
1686 : *size, communicator->GetId().c_str());
1687 : }
1688 2 : return HCCL_SUCCESS;
1689 2 : }
1690 :
1691 1 : HcclResult HcclGetHcclBufferV2(HcclComm comm, void** addr, uint64_t* size)
1692 : {
1693 1 : CHK_PTR_NULL(comm);
1694 1 : CHK_PTR_NULL(addr);
1695 1 : CHK_PTR_NULL(size);
1696 1 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
1697 1 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1698 0 : HCCL_RUN_INFO("Entry-HcclGetHcclBuffer V950, commId[%s]", communicator->GetId().c_str());
1699 : }
1700 1 : CHK_RET(communicator->GetLocalCclBuffer(addr, size));
1701 : /* 关键状态记录 */
1702 1 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1703 0 : HCCL_RUN_INFO(
1704 : "HcclGetHcclBuffer success, addr[%p], size[%llu], commId[%s]", addr, *size, communicator->GetId().c_str());
1705 : }
1706 1 : return HCCL_SUCCESS;
1707 : }
1708 :
1709 1 : HcclResult HcclGetRemoteIpcHcclBufV2(HcclComm comm, uint64_t remoteRank, void** addr, uint64_t* size)
1710 : {
1711 1 : CHK_PTR_NULL(comm);
1712 1 : CHK_PTR_NULL(addr);
1713 1 : CHK_PTR_NULL(size);
1714 1 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
1715 1 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1716 0 : HCCL_RUN_INFO(
1717 : "Entry-HcclGetRemoteIpcHcclBuf V950 start, remoteRank[%llu], addr[%p], size[%llu], commId[%s]", remoteRank,
1718 : *addr, *size, communicator->GetId().c_str());
1719 : }
1720 3 : HCCL_ERROR("Entry-HcclGetRemoteIpcHcclBuf V950 not support, commId[%s]", communicator->GetId().c_str());
1721 1 : return HCCL_E_NOT_SUPPORT;
1722 : }
1723 :
1724 1 : HcclResult HcclGetAicpuOpStreamAndNotifyV2(HcclComm comm, rtStream_t* opstream, u8 aicpuNotifyNum, void** aicpuNotify)
1725 : {
1726 1 : CHK_PTR_NULL(comm);
1727 1 : CHK_PTR_NULL(opstream);
1728 1 : CHK_PTR_NULL(aicpuNotify);
1729 1 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
1730 1 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
1731 0 : HCCL_RUN_INFO(
1732 : "Entry-HcclGetAicpuOpStreamAndNotify V950, aicpuNotifyNum[%u], commId[%s]", aicpuNotifyNum,
1733 : communicator->GetId().c_str());
1734 : }
1735 1 : CHK_RET(communicator->GetAicpuOpStreamNotify(opstream, aicpuNotifyNum, aicpuNotify));
1736 1 : return HCCL_SUCCESS;
1737 : }
1738 :
1739 2 : HcclResult HcclCommResumeV2(HcclComm comm)
1740 : {
1741 2 : CHK_PTR_NULL(comm);
1742 2 : HcclUs startut = TIME_NOW();
1743 6 : HCCL_RUN_INFO("Entry-HcclCommResume V950");
1744 2 : s32 deviceLogicId = HcclGetThreadDeviceId();
1745 2 : s32 devPhyId = HrtGetDevicePhyIdByIndex(deviceLogicId);
1746 2 : CHK_RET(static_cast<HcclResult>(Hccl::HcclCcuResumePfeTableProcess(deviceLogicId)));
1747 2 : CHK_RET(HcclCommResumeImplV2(comm));
1748 6 : HCCL_RUN_INFO(
1749 : "Entry-HcclCommResume V950 success, deviceLogicId[%d], devPhyId[%d], take time [%lld]us", deviceLogicId,
1750 : devPhyId, DURATION_US(TIME_NOW() - startut));
1751 2 : return HCCL_SUCCESS;
1752 : }
1753 :
1754 2 : HcclResult HcclCommResumeImplV2(HcclComm comm)
1755 : {
1756 2 : CHK_PTR_NULL(comm);
1757 6 : return HcclCommOperationImplV2(comm, "HcclCommResumeV2", [](Hccl::HcclCommunicator& communicator) {
1758 6 : HCCL_RUN_INFO("Entry-HcclCommResume commId[%s]", communicator.GetId().c_str());
1759 2 : return static_cast<HcclResult>(communicator.Resume());
1760 2 : });
1761 : }
1762 :
1763 5 : HcclResult RootInfoDetect(
1764 : const u32 nRanks, u32 rank, const HcclRootHandleV2& rootHandle, RankTableInfo& rankTable,
1765 : RootInfoDetectBridge::DetectContext& detectContext)
1766 : {
1767 5 : const RootInfoDetectBridge* bridge = GetRootInfoDetectBridge();
1768 5 : CHK_PRT_RET(
1769 : bridge == nullptr || bridge->detectRankTable == nullptr,
1770 : HCCL_ERROR("[%s] RootInfoDetect bridge is not registered.", __func__), HCCL_E_INTERNAL);
1771 5 : return bridge->detectRankTable(nRanks, rank, rootHandle, rankTable, detectContext);
1772 : }
1773 :
1774 : HcclResult
1775 3 : CommInitRootInfo(u32 nRanks, u32 rank, const HcclRootHandleV2& rootHandle, const string& identifier, HcclComm* comm)
1776 : {
1777 : // 临时规避，在初始化通信域前声明单例保证时序
1778 3 : CHK_RET(CallSingletons());
1779 : // check
1780 3 : HcclCommInfoV2& opbasedCommInfoV2 = GetCommInfoV2();
1781 6 : CHK_PRT_RET(
1782 : opbasedCommInfoV2.hcclGroupMap.find(identifier) != opbasedCommInfoV2.hcclGroupMap.end(),
1783 : HCCL_ERROR(
1784 : "[CreateCommConfig] errNo[0x%016llx] The rootHandle[%s] already exists in Group2Comm map.",
1785 : HCCL_ERROR_CODE(HCCL_E_PARA), identifier.c_str()),
1786 : HCCL_E_PARA);
1787 :
1788 : // rootInfo获取rankTable, 基于rankTable创建通信域
1789 2 : RootInfoDetectBridge::DetectContext rootInfoDetectContext;
1790 2 : RankTableInfo rankTable{};
1791 2 : HcclResult ret = RootInfoDetect(nRanks, rank, rootHandle, rankTable, rootInfoDetectContext);
1792 2 : if (ret != HCCL_SUCCESS) {
1793 0 : HCCL_ERROR(
1794 : "[%s] errNo[0x%016llx] RootInfoDetect failed, rootHandle[%s].", __func__, HCCL_ERROR_CODE(ret),
1795 : identifier.c_str());
1796 0 : rankTable.Dump();
1797 0 : return ret;
1798 : }
1799 :
1800 : // 打印ranktable
1801 2 : rankTable.Dump();
1802 :
1803 : // 创建通信域
1804 2 : bool devUsed = false;
1805 2 : bool isWorldGroup = true;
1806 : Hccl::CommParams commParams{
1807 : identifier,
1808 : static_cast<Hccl::RankId>(rank),
1809 : nRanks,
1810 : static_cast<Hccl::RankId>(rank),
1811 0 : Hccl::HrtGetDeviceType(),
1812 : devUsed,
1813 2 : isWorldGroup};
1814 2 : opbasedCommInfoV2.pComm.reset(new (std::nothrow) Hccl::HcclCommunicator(commParams));
1815 2 : opbasedCommInfoV2.commParams = commParams;
1816 :
1817 : // 通信域初始化
1818 2 : CHK_PTR_NULL(opbasedCommInfoV2.pComm);
1819 : /* --------------初始化------------------------- */
1820 2 : bool errorFlag = false;
1821 2 : s32 logicDevId = HrtGetDevice();
1822 : do {
1823 2 : ret = opbasedCommInfoV2.pComm->Init(rankTable);
1824 2 : CHK_PRT_BREAK(
1825 : ret != HcclResult::HCCL_SUCCESS,
1826 : HCCL_ERROR("[%s]opbasedCommInfoV2.pComm->Init failed, errNo[0x%016llx]", __func__, HCCL_ERROR_CODE(ret)),
1827 : errorFlag = true);
1828 : // 配置默认加速模式
1829 2 : opbasedCommInfoV2.pComm->RegisterAcceStateCallBack(CommunicatorCallback());
1830 2 : ret = CommManager::GetInstance(logicDevId)
1831 2 : .SetCommAcceleratorV2(opbasedCommInfoV2.pComm.get(), 0); // 通信域创建，设置默认accelerator
1832 2 : CHK_PRT_BREAK(
1833 : ret != HcclResult::HCCL_SUCCESS,
1834 : HCCL_ERROR("[%s]SetCommAcceleratorV2 failed, errNo[0x%016llx]", __func__, HCCL_ERROR_CODE(ret)),
1835 : errorFlag = true);
1836 : // 保存通信域
1837 2 : HcclGroupParamsV2 params{};
1838 2 : params.pComm = opbasedCommInfoV2.pComm;
1839 2 : params.groupRank = static_cast<Hccl::RankId>(rank);
1840 2 : std::unique_lock<std::mutex> lock(opbasedCommInfoV2.groupParamsLock);
1841 2 : opbasedCommInfoV2.hcclGroupMap[identifier] = params;
1842 :
1843 4 : opbasedCommInfoV2.pComm->RegisterPrintChannelInfoCallback(
1844 4 : CommManager::GetInstance(logicDevId).GetPrintChannelInfoCallback());
1845 :
1846 2 : *comm = static_cast<HcclComm>(opbasedCommInfoV2.pComm.get());
1847 2 : } while (0);
1848 :
1849 2 : if (errorFlag) {
1850 0 : HCCL_ERROR(
1851 : "[Init][%s]HcclCommInitClusterInfoV2 failed, rankNum[%u], rank[%u], logicDevId[%d], rootInfo "
1852 : "identifier[%s],"
1853 : "return[0x%016llx]",
1854 : __func__, nRanks, rank, logicDevId, identifier.c_str(), HCCL_ERROR_CODE(ret));
1855 0 : (void)HcclCommDestroyV2(opbasedCommInfoV2.pComm.get());
1856 0 : *comm = nullptr;
1857 0 : return ret;
1858 : }
1859 6 : HCCL_INFO(
1860 : "[%s] Init success, rankNum[%u], rank[%u], rootInfo identifier[%s], logicDevId[%d]", __func__, nRanks, rank,
1861 : identifier.c_str(), logicDevId);
1862 :
1863 2 : return HCCL_SUCCESS;
1864 2 : }
1865 :
1866 3 : HcclResult HcclCommInitRootInfoV2(
1867 : uint32_t nRanks, const HcclRootInfo* rootInfo, uint32_t rank, HcclComm* comm, std::string& identifier)
1868 : {
1869 3 : HcclUs startut = TIME_NOW();
1870 3 : CHK_PTR_NULL(rootInfo);
1871 9 : HCCL_RUN_INFO("Entry-HcclCommInitRootInfo V950, rankId[%u], rankNum[%u].", rank, nRanks);
1872 :
1873 : // 获取rootHandle
1874 3 : HcclRootHandleV2 rootHandle{};
1875 3 : s32 sRet = memcpy_s(&rootHandle, sizeof(HcclRootHandleV2), rootInfo->internal, sizeof(HcclRootHandleV2));
1876 3 : CHK_PRT_RET(
1877 : sRet != EOK,
1878 : HCCL_ERROR("[%s] memcpy root info fail. errorno[%d] length[%u]", __func__, sRet, sizeof(rootHandle)),
1879 : HCCL_E_MEMORY);
1880 :
1881 : // 获取通信域name
1882 3 : rootHandle.identifier[ROOTINFO_INDENTIFIER_MAX_LENGTH - 1] = '\0';
1883 3 : identifier = rootHandle.identifier;
1884 :
1885 : /* 接口交互信息日志 */
1886 3 : s32 deviceLogicId = HcclGetThreadDeviceId();
1887 3 : s32 devPhyId = HrtGetDevicePhyIdByIndex(deviceLogicId);
1888 9 : HCCL_RUN_INFO(
1889 : "Entry-HcclCommInitRootInfo V950, ranks[%u], rank[%u], rootinfo: host ip[%s] port[%u] "
1890 : "netMode[%s] identifier[%s], deviceLogicId[%d], devPhyId[%d]",
1891 : nRanks, rank, rootHandle.ip, rootHandle.listenPort, rootHandle.netMode.Describe().c_str(), identifier.c_str(),
1892 : deviceLogicId, devPhyId);
1893 :
1894 : // rootInfo获取rankTable, 基于rankTable创建通信域
1895 3 : HcclResult ret = CommInitRootInfo(nRanks, rank, rootHandle, identifier, comm);
1896 6 : CHK_PRT_RET(
1897 : ret != HCCL_SUCCESS,
1898 : HCCL_ERROR(
1899 : "[%s] errNo[0x%016llx] CommInitRootInfo failed, identifier[%s].", __func__, HCCL_ERROR_CODE(ret),
1900 : identifier.c_str()),
1901 : ret);
1902 :
1903 : /* 关键状态记录 */
1904 6 : HCCL_RUN_INFO(
1905 : "HcclCommInitRootInfoV2 success, take time [%lld]us, rankNum[%u], rank[%u], identifier[%s]",
1906 : DURATION_US(TIME_NOW() - startut), nRanks, rank, identifier.c_str());
1907 2 : return HCCL_SUCCESS;
1908 : }
1909 :
1910 3 : HcclResult HcclCommInitRootInfoConfigV2(
1911 : uint32_t nRanks, const HcclRootInfo* rootInfo, uint32_t rank, const HcclCommConfig* config, HcclComm* comm)
1912 : {
1913 3 : HcclUs startut = TIME_NOW();
1914 3 : CHK_PTR_NULL(rootInfo);
1915 3 : CHK_PTR_NULL(config);
1916 9 : HCCL_RUN_INFO(
1917 : "Entry-HcclCommInitRootInfoConfig V950: nRanks[%u], rank[%u], commEngine[%s]", nRanks, rank,
1918 : HcclAccelerator(static_cast<HcclAccelerator::Value>(config->hcclOpExpansionMode)).Describe().c_str());
1919 : // 获取rootHandle
1920 3 : HcclRootHandleV2 rootHandle{};
1921 3 : s32 sRet = memcpy_s(&rootHandle, sizeof(rootHandle), rootInfo->internal, sizeof(rootHandle));
1922 3 : CHK_PRT_RET(
1923 : sRet != EOK,
1924 : HCCL_ERROR("[%s]memcpy root info fail. errorno[%d] count[%u]", __func__, sRet, sizeof(HcclRootHandleV2)),
1925 : HCCL_E_MEMORY);
1926 :
1927 : // 获取通信域name
1928 3 : rootHandle.identifier[ROOTINFO_INDENTIFIER_MAX_LENGTH - 1] = '\0';
1929 3 : string identifier = strlen(config->hcclCommName) != 0 ? config->hcclCommName : rootHandle.identifier;
1930 :
1931 : /* 接口交互信息日志 */
1932 9 : HCCL_RUN_INFO(
1933 : "Entry-HcclCommInitRootInfoConfigV2:ranks[%u], rank[%u], rootinfo: host ip[%s] port[%u] "
1934 : "netMode[%s] rootHandle.identifier[%s], identifier[%s]",
1935 : nRanks, rank, rootHandle.ip, rootHandle.listenPort, rootHandle.netMode.Describe().c_str(),
1936 : rootHandle.identifier, identifier.c_str());
1937 :
1938 : // 临时规避，在初始化通信域前声明单例保证时序
1939 3 : CHK_RET(CallSingletons());
1940 3 : HcclCommInfoV2& opbasedCommInfoV2 = GetCommInfoV2();
1941 3 : CHK_PRT_RET(
1942 : opbasedCommInfoV2.hcclGroupMap.find(identifier) != opbasedCommInfoV2.hcclGroupMap.end(),
1943 : HCCL_ERROR(
1944 : "[HcclCommInitRootInfoConfigV2]errNo[0x%016llx] The comm name[%s] already exists in Group2Comm map.",
1945 : HCCL_ERROR_CODE(HCCL_E_PARA), identifier.c_str()),
1946 : HCCL_E_PARA);
1947 :
1948 3 : RankTableInfo rankTable{};
1949 3 : RootInfoDetectBridge::DetectContext rootInfoDetectContext;
1950 3 : HcclResult ret = RootInfoDetect(nRanks, rank, rootHandle, rankTable, rootInfoDetectContext);
1951 3 : if (ret != HCCL_SUCCESS) {
1952 3 : HCCL_ERROR(
1953 : "[%s] errNo[0x%016llx] RootInfoDetect failed, identifier[%s].", __func__, HCCL_ERROR_CODE(ret),
1954 : identifier.c_str());
1955 1 : rankTable.Dump();
1956 1 : return ret;
1957 : }
1958 :
1959 : // 打印ranktable
1960 2 : rankTable.Dump();
1961 :
1962 : // 创建通信域
1963 2 : ret = CreateCommConfigRootInfo(rank, config, identifier, rankTable, comm);
1964 2 : CHK_PRT_RET(
1965 : ret,
1966 : HCCL_ERROR(
1967 : "[%s]errNo[0x%016llx] and create comm failed, identifier[%s].", __func__, HCCL_ERROR_CODE(ret),
1968 : identifier.c_str()),
1969 : static_cast<HcclResult>(ret));
1970 :
1971 : /* 关键状态记录 */
1972 6 : HCCL_RUN_INFO(
1973 : "HcclCommInitRootInfoConfigV2 success, take time [%lld]us, rankNum[%u], rank[%u], identifier[%s]",
1974 : DURATION_US(TIME_NOW() - startut), nRanks, rank, identifier.c_str());
1975 2 : return HCCL_SUCCESS;
1976 3 : }
1977 :
1978 2 : HcclResult HcclScatterV2(
1979 : void* sendBuf, void* recvBuf, uint64_t recvCount, HcclDataType dataType, uint32_t root, HcclComm comm,
1980 : aclrtStream stream)
1981 : {
1982 2 : HcclUs startut = TIME_NOW();
1983 : bool isCapture;
1984 2 : rtModel_t rtModel = nullptr;
1985 2 : u32 modelId = 0;
1986 :
1987 2 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
1988 2 : const std::string tag = "Scatter_" + communicator->GetId();
1989 :
1990 2 : CHK_RET_AND_PRINT_IDE(HcomCheckOpParamV2(tag.c_str(), recvCount, dataType, stream), tag.c_str());
1991 2 : u32 rankSize = INVALID_VALUE_RANKSIZE;
1992 2 : CHK_RET_AND_PRINT_IDE(communicator->GetRankSize(&rankSize), tag.c_str());
1993 2 : CHK_RET_AND_PRINT_IDE(HcomCheckUserRankV2(rankSize, root), tag.c_str());
1994 2 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
1995 :
1996 2 : u32 rankId = INVALID_VALUE_RANKID;
1997 2 : CHK_RET(communicator->GetRankId(rankId));
1998 2 : if (rankId == root) { // 本rank为root节点，send_buff不为空
1999 2 : CHK_PTR_NULL(sendBuf);
2000 : }
2001 :
2002 : /* 接口交互信息日志 */
2003 : char stackLogBufferV2[LOG_TMPBUF_SIZE];
2004 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2005 1 : s32 streamId = HrtGetStreamId(stream);
2006 1 : s32 deviceLogicId = HrtGetDevice();
2007 1 : u32 localRank = INVALID_VALUE_RANKID;
2008 1 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
2009 :
2010 1 : s32 ret = snprintf_s(
2011 : stackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
2012 : "tag[%s], sendBuf[%p], recvBuf[%p], recvCount[%llu], dataType[%s], root[%u], localRank[%u], streamId[%d], "
2013 : "deviceLogicId[%d]",
2014 2 : tag.c_str(), sendBuf, recvBuf, recvCount, GetDataTypeEnumStrV2(dataType).c_str(), root, localRank, streamId,
2015 : deviceLogicId);
2016 :
2017 1 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
2018 1 : std::string logInfo = "Entry-HcclScatterV2:" + std::string(stackLogBufferV2);
2019 1 : if (isCapture) {
2020 0 : CHK_PTR_NULL(rtModel);
2021 : // 获取不到modelId会报错
2022 0 : CHK_RET(GetModelId(rtModel, modelId));
2023 0 : logInfo += ", model id[" + to_string(modelId) + "].";
2024 : }
2025 1 : communicator->GetTrace().Save(logInfo);
2026 1 : }
2027 :
2028 2 : static thread_local Hccl::CollOpParams opParams;
2029 2 : opParams.opType = Hccl::OpType::SCATTER;
2030 2 : opParams.dataType = HcclDataTypeToDataType(dataType);
2031 2 : opParams.sendBuf = sendBuf;
2032 2 : opParams.recvBuf = recvBuf;
2033 2 : opParams.count = recvCount;
2034 2 : opParams.root = root;
2035 :
2036 2 : CHK_RET_AND_PRINT_IDE(communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream)), tag.c_str());
2037 :
2038 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2039 1 : HcclUs endut = TIME_NOW();
2040 : /* 关键状态记录 */
2041 2 : std::string endInfo = "HcclScatterV2:success,take time: " + std::to_string(DURATION_US(endut - startut).count())
2042 3 : + " us, tag: " + tag + std::string(stackLogBufferV2);
2043 1 : communicator->GetTrace().Save(endInfo);
2044 1 : }
2045 :
2046 2 : return HCCL_SUCCESS;
2047 2 : }
2048 :
2049 3 : HcclResult HcclAllGatherV2(
2050 : void* sendBuf, void* recvBuf, uint64_t sendCount, HcclDataType dataType, HcclComm comm, aclrtStream stream)
2051 : {
2052 3 : HcclUs startut = TIME_NOW();
2053 : bool isCapture;
2054 3 : rtModel_t rtModel = nullptr;
2055 3 : u32 modelId = 0;
2056 :
2057 3 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
2058 3 : const std::string tag = "AllGather_" + communicator->GetId();
2059 :
2060 3 : CHK_RET_AND_PRINT_IDE(HcomCheckOpParamV2(tag.c_str(), sendCount, dataType, stream), tag.c_str());
2061 3 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
2062 :
2063 : /* 接口交互信息日志 */
2064 : char stackLogBufferV2[LOG_TMPBUF_SIZE];
2065 3 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2066 1 : s32 streamId = HrtGetStreamId(stream);
2067 1 : s32 deviceLogicId = HrtGetDevice();
2068 1 : u32 localRank = INVALID_VALUE_RANKID;
2069 1 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
2070 :
2071 1 : s32 ret = snprintf_s(
2072 : stackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
2073 : "tag[%s], sendBuf[%p], recvBuf[%p], sendCount[%llu], dataType[%s], localRank[%u], streamId[%d],"
2074 : "deviceLogicId[%d]",
2075 2 : tag.c_str(), sendBuf, recvBuf, sendCount, GetDataTypeEnumStrV2(dataType).c_str(), localRank, streamId,
2076 : deviceLogicId);
2077 :
2078 1 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
2079 1 : std::string logInfo = "Entry-HcclAllGatherV2:" + std::string(stackLogBufferV2);
2080 1 : if (isCapture) {
2081 0 : CHK_PTR_NULL(rtModel);
2082 : // 获取不到modelId会报错
2083 0 : CHK_RET(GetModelId(rtModel, modelId));
2084 0 : logInfo += ", model id[" + to_string(modelId) + "].";
2085 : }
2086 1 : communicator->GetTrace().Save(logInfo);
2087 1 : }
2088 :
2089 3 : static thread_local Hccl::CollOpParams opParams;
2090 3 : opParams.opType = Hccl::OpType::ALLGATHER;
2091 3 : opParams.dataType = HcclDataTypeToDataType(dataType);
2092 3 : opParams.sendBuf = sendBuf;
2093 3 : opParams.recvBuf = recvBuf;
2094 3 : opParams.count = sendCount;
2095 3 : opParams.opTag = tag;
2096 :
2097 3 : CHK_RET_AND_PRINT_IDE(communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream)), tag.c_str());
2098 :
2099 3 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2100 1 : HcclUs endut = TIME_NOW();
2101 : /* 关键状态记录 */
2102 : std::string endInfo
2103 2 : = "HcclAllGatherV2:success,take time: " + std::to_string(DURATION_US(endut - startut).count())
2104 3 : + " us, tag: " + tag + std::string(stackLogBufferV2);
2105 1 : communicator->GetTrace().Save(endInfo);
2106 1 : }
2107 :
2108 3 : return HCCL_SUCCESS;
2109 3 : }
2110 :
2111 9 : HcclResult HcclAllGatherVV2(
2112 : void* sendBuf, uint64_t sendCount, void* recvBuf, void* recvCounts, void* recvDispls, HcclDataType dataType,
2113 : HcclComm comm, aclrtStream stream)
2114 : {
2115 9 : HcclUs startut = TIME_NOW();
2116 : bool isCapture;
2117 9 : rtModel_t rtModel = nullptr;
2118 9 : u32 modelId = 0;
2119 :
2120 : // 获取通信域
2121 9 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
2122 9 : const std::string tag = "AllGatherV_" + communicator->GetId();
2123 :
2124 : // 获取rank信息
2125 : uint32_t rankId;
2126 9 : CHK_RET(communicator->GetRankId(rankId));
2127 : uint32_t rankSize;
2128 9 : CHK_RET(communicator->GetRankSize(&rankSize));
2129 : // 参数合法性校验
2130 9 : if (rankSize == 1) {
2131 : // rankSize为1时，退化为AllGather
2132 : // 检查异常回退AGV的情况
2133 3 : if (sendCount == 0) {
2134 3 : HCCL_WARNING("[AllGatherV] sendCount is 0 when single rank");
2135 1 : return HCCL_SUCCESS;
2136 : } else {
2137 5 : CHK_PRT_RET(sendBuf == nullptr, HCCL_ERROR("[AllGatherV] sendBuf is null when single rank"), HCCL_E_PTR);
2138 1 : CHK_PRT_RET(recvBuf == nullptr, HCCL_ERROR("[AllGatherV] recvBuf is null when single rank"), HCCL_E_PTR);
2139 : }
2140 1 : return HcclAllGatherV2(sendBuf, recvBuf, sendCount, dataType, comm, stream);
2141 : }
2142 18 : CHK_RET_AND_PRINT_IDE(HcomCheckOpParamV2(tag.c_str(), sendCount, dataType, stream), tag.c_str());
2143 10 : CHK_RET_AND_PRINT_IDE(HcomCheckVOpParamV2(rankId, rankSize, sendCount, recvCounts), tag.c_str());
2144 3 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
2145 :
2146 : /* 接口交互信息日志 */
2147 : char stackLogBufferV2[LOG_TMPBUF_SIZE];
2148 3 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2149 0 : s32 streamId = HrtGetStreamId(stream);
2150 0 : s32 deviceLogicId = HrtGetDevice();
2151 0 : u32 localRank = INVALID_VALUE_RANKID;
2152 0 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
2153 :
2154 0 : s32 ret = snprintf_s(
2155 : stackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
2156 : "tag[%s], sendBuf[%p], recvBuf[%p], sendCount[%llu], recvCounts[%p], recvDispls[%p], "
2157 : "dataType[%s], localRank[%u], streamId[%d], deviceLogicId[%d]",
2158 0 : tag.c_str(), sendBuf, recvBuf, sendCount, recvCounts, recvDispls, GetDataTypeEnumStrV2(dataType).c_str(),
2159 : localRank, streamId, deviceLogicId);
2160 :
2161 0 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
2162 0 : std::string logInfo = "Entry-HcclAllGatherVV2:" + std::string(stackLogBufferV2);
2163 0 : if (isCapture) {
2164 0 : CHK_PTR_NULL(rtModel);
2165 : // 获取不到modelId会报错
2166 0 : CHK_RET(GetModelId(rtModel, modelId));
2167 0 : logInfo += ", model id[" + to_string(modelId) + "].";
2168 : }
2169 0 : communicator->GetTrace().Save(logInfo);
2170 0 : }
2171 :
2172 3 : u64* counts = static_cast<u64*>(recvCounts);
2173 3 : u64 output = 0;
2174 9 : for (size_t index = 0; index < rankSize; index++) {
2175 6 : output += counts[index];
2176 : }
2177 3 : if (output == 0) {
2178 3 : HCCL_INFO("[%s] output[%llu] is equal to zero", __func__, output);
2179 1 : return HCCL_SUCCESS;
2180 : }
2181 2 : RPT_INPUT_ERR(
2182 : recvBuf == nullptr, "EI0003", std::vector<std::string>({"ccl_op", "value", "parameter", "expect"}),
2183 : std::vector<std::string>({"HcclAllGatherVV2", "nullptr", "recvBuf", "not nullptr"}));
2184 2 : CHK_PTR_NULL(recvBuf);
2185 : // opParams组装
2186 2 : Hccl::CollOpParams opParams;
2187 2 : opParams.opType = Hccl::OpType::ALLGATHERV;
2188 2 : opParams.dataType = HcclDataTypeToDataType(dataType);
2189 2 : opParams.dstRank = rankId;
2190 2 : opParams.sendBuf = sendBuf;
2191 2 : opParams.recvBuf = recvBuf;
2192 2 : opParams.count = sendCount;
2193 2 : opParams.vDataDes.counts = recvCounts;
2194 2 : opParams.vDataDes.displs = recvDispls;
2195 2 : opParams.vDataDes.dataType = HcclDataTypeToDataType(dataType);
2196 :
2197 2 : CHK_RET_AND_PRINT_IDE(communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream)), tag.c_str());
2198 :
2199 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2200 0 : HcclUs endut = TIME_NOW();
2201 : /* 关键状态记录 */
2202 : std::string endInfo
2203 0 : = "HcclAllGatherVV2:success,take time: " + std::to_string(DURATION_US(endut - startut).count())
2204 0 : + " us, tag: " + tag + std::string(stackLogBufferV2);
2205 0 : communicator->GetTrace().Save(endInfo);
2206 0 : }
2207 :
2208 2 : return HCCL_SUCCESS;
2209 9 : }
2210 :
2211 4 : HcclResult ValidateRank(uint32_t rank, Hccl::HcclCommunicator* communicator)
2212 : {
2213 4 : u32 rankSize{};
2214 4 : CHK_RET(communicator->GetRankSize(&rankSize));
2215 4 : u32 rankId{INVALID_VALUE_RANKID};
2216 4 : CHK_RET(communicator->GetRankId(rankId));
2217 4 : CHK_RET(HcomCheckUserRankV2(rankSize, rank));
2218 4 : CHK_PRT_RET(rankId == rank, HCCL_ERROR("same rank[%u] is not allowed", rank), HCCL_E_PARA);
2219 4 : return HCCL_SUCCESS;
2220 : }
2221 :
2222 : HcclResult
2223 2 : HcclSendV2(void* sendBuf, uint64_t count, HcclDataType dataType, uint32_t destRank, HcclComm comm, aclrtStream stream)
2224 : {
2225 2 : HcclUs startut = TIME_NOW();
2226 : bool isCapture;
2227 2 : rtModel_t rtModel = nullptr;
2228 2 : u32 modelId = 0;
2229 :
2230 2 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
2231 2 : const std::string tag = "SendRecv_" + communicator->GetId();
2232 :
2233 2 : CHK_RET(HcomCheckDataTypeV2(dataType));
2234 2 : CHK_RET(ValidateRank(destRank, communicator));
2235 2 : CHK_RET_AND_PRINT_IDE(HcomCheckOpParamV2(tag.c_str(), count, dataType, stream), tag.c_str());
2236 2 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
2237 :
2238 : /* 接口交互信息日志 */
2239 : char hcclSendStackLogBufferV2[LOG_TMPBUF_SIZE];
2240 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2241 1 : s32 streamId = HrtGetStreamId(stream);
2242 1 : s32 deviceLogicId = HrtGetDevice();
2243 1 : u32 localRank = INVALID_VALUE_RANKID;
2244 1 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
2245 :
2246 1 : s32 ret = snprintf_s(
2247 : hcclSendStackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
2248 : "tag[%s], sendBuf[%p], count[%llu], dataType[%s], localRank[%u], streamId[%d], deviceLogicId[%d]",
2249 2 : tag.c_str(), sendBuf, count, GetDataTypeEnumStrV2(dataType).c_str(), localRank, streamId, deviceLogicId);
2250 :
2251 1 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
2252 1 : std::string logInfo = "Entry-HcclSendV2:" + std::string(hcclSendStackLogBufferV2);
2253 1 : if (isCapture) {
2254 0 : CHK_PTR_NULL(rtModel);
2255 : // 获取不到modelId会报错
2256 0 : CHK_RET(GetModelId(rtModel, modelId));
2257 0 : logInfo += ", model id[" + to_string(modelId) + "].";
2258 : }
2259 1 : communicator->GetTrace().Save(logInfo);
2260 1 : }
2261 :
2262 2 : static thread_local Hccl::CollOpParams opParams{};
2263 2 : opParams.opType = Hccl::OpType::SEND;
2264 2 : opParams.dataType = HcclDataTypeToDataType(dataType);
2265 2 : opParams.sendBuf = sendBuf;
2266 2 : opParams.recvBuf = nullptr;
2267 2 : opParams.count = count;
2268 2 : opParams.dstRank = destRank;
2269 2 : opParams.opTag = tag;
2270 :
2271 2 : CHK_RET_AND_PRINT_IDE(communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream)), tag.c_str());
2272 :
2273 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2274 1 : HcclUs endut = TIME_NOW();
2275 : /* 关键状态记录 */
2276 2 : std::string endInfo = "HcclSendV2:success,take time: " + std::to_string(DURATION_US(endut - startut).count())
2277 3 : + " us, tag: " + tag + std::string(hcclSendStackLogBufferV2);
2278 1 : communicator->GetTrace().Save(endInfo);
2279 1 : }
2280 :
2281 2 : return HCCL_SUCCESS;
2282 2 : }
2283 :
2284 : HcclResult
2285 2 : HcclRecvV2(void* recvBuf, uint64_t count, HcclDataType dataType, uint32_t srcRank, HcclComm comm, aclrtStream stream)
2286 : {
2287 2 : HcclUs startut = TIME_NOW();
2288 : bool isCapture;
2289 2 : rtModel_t rtModel = nullptr;
2290 2 : u32 modelId = 0;
2291 :
2292 2 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
2293 2 : const std::string tag = "SendRecv_" + communicator->GetId();
2294 :
2295 2 : CHK_RET(HcomCheckDataTypeV2(dataType));
2296 2 : CHK_RET(ValidateRank(srcRank, communicator));
2297 2 : CHK_RET_AND_PRINT_IDE(HcomCheckOpParamV2(tag.c_str(), count, dataType, stream), tag.c_str());
2298 2 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
2299 :
2300 : /* 接口交互信息日志 */
2301 : char hcclRecvStackLogBufferV2[LOG_TMPBUF_SIZE];
2302 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2303 1 : s32 streamId = HrtGetStreamId(stream);
2304 1 : s32 deviceLogicId = HrtGetDevice();
2305 1 : u32 localRank = INVALID_VALUE_RANKID;
2306 1 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
2307 :
2308 1 : s32 ret = snprintf_s(
2309 : hcclRecvStackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
2310 : "tag[%s], recvBuf[%p], count[%llu], dataType[%s], localRank[%u], streamId[%d], deviceLogicId[%d]",
2311 2 : tag.c_str(), recvBuf, count, GetDataTypeEnumStrV2(dataType).c_str(), localRank, streamId, deviceLogicId);
2312 :
2313 1 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
2314 1 : std::string logInfo = "Entry-HcclRecvV2:" + std::string(hcclRecvStackLogBufferV2);
2315 1 : if (isCapture) {
2316 0 : CHK_PTR_NULL(rtModel);
2317 : // 获取不到modelId会报错
2318 0 : CHK_RET(GetModelId(rtModel, modelId));
2319 0 : logInfo += ", model id[" + to_string(modelId) + "].";
2320 : }
2321 1 : communicator->GetTrace().Save(logInfo);
2322 1 : }
2323 :
2324 2 : static thread_local Hccl::CollOpParams opParams{};
2325 2 : opParams.opType = Hccl::OpType::RECV;
2326 2 : opParams.dataType = HcclDataTypeToDataType(dataType);
2327 2 : opParams.sendBuf = nullptr;
2328 2 : opParams.recvBuf = recvBuf;
2329 2 : opParams.count = count;
2330 2 : opParams.dstRank = srcRank;
2331 2 : opParams.opTag = tag;
2332 :
2333 2 : CHK_RET_AND_PRINT_IDE(communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream)), tag.c_str());
2334 :
2335 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2336 1 : HcclUs endut = TIME_NOW();
2337 : /* 关键状态记录 */
2338 2 : std::string endInfo = "HcclRecvV2:success,take time: " + std::to_string(DURATION_US(endut - startut).count())
2339 3 : + " us, tag: " + tag + std::string(hcclRecvStackLogBufferV2);
2340 1 : communicator->GetTrace().Save(endInfo);
2341 1 : }
2342 :
2343 2 : return HCCL_SUCCESS;
2344 2 : }
2345 :
2346 9 : HcclResult HcclReduceScatterV2(
2347 : void* sendBuf, void* recvBuf, uint64_t recvCount, HcclDataType dataType, HcclReduceOp op, HcclComm comm,
2348 : aclrtStream stream)
2349 : {
2350 9 : HcclUs startut = TIME_NOW();
2351 : bool isCapture;
2352 9 : rtModel_t rtModel = nullptr;
2353 9 : u32 modelId = 0;
2354 :
2355 9 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
2356 9 : const std::string tag = "ReduceScatter_" + communicator->GetId();
2357 :
2358 9 : CHK_RET_AND_PRINT_IDE(HcomCheckOpParamV2(tag.c_str(), recvCount, dataType, stream), tag.c_str());
2359 9 : CHK_RET_AND_PRINT_IDE(HcomCheckReductionOpV2(op), tag.c_str());
2360 21 : CHK_RET_AND_PRINT_IDE(HcomCheckReduceDataTypeV2(dataType, op), tag.c_str());
2361 7 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
2362 :
2363 : /* 接口交互信息日志 */
2364 : char stackLogBufferV2[LOG_TMPBUF_SIZE];
2365 7 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2366 3 : s32 streamId = HrtGetStreamId(stream);
2367 3 : s32 deviceLogicId = HrtGetDevice();
2368 3 : u32 localRank = INVALID_VALUE_RANKID;
2369 3 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
2370 :
2371 6 : s32 ret = snprintf_s(
2372 : stackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
2373 : "tag[%s], sendBuf[%p], recvBuf[%p], recvCount[%llu], dataType[%s], op[%s],"
2374 : "localRank[%u], streamId[%d], deviceLogicId[%d]",
2375 6 : tag.c_str(), sendBuf, recvBuf, recvCount, GetDataTypeEnumStrV2(dataType).c_str(),
2376 6 : GetReduceOpEnumStrV2(op).c_str(), localRank, streamId, deviceLogicId);
2377 :
2378 3 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
2379 3 : std::string logInfo = "Entry-HcclReduceScatterV2:" + std::string(stackLogBufferV2);
2380 3 : if (isCapture) {
2381 0 : CHK_PTR_NULL(rtModel);
2382 : // 获取不到modelId会报错
2383 0 : CHK_RET(GetModelId(rtModel, modelId));
2384 0 : logInfo += ", model id[" + to_string(modelId) + "].";
2385 : }
2386 3 : communicator->GetTrace().Save(logInfo);
2387 3 : }
2388 :
2389 7 : static thread_local Hccl::CollOpParams opParams;
2390 7 : opParams.opType = Hccl::OpType::REDUCESCATTER;
2391 7 : opParams.dataType = HcclDataTypeToDataType(dataType);
2392 7 : opParams.reduceOp = HCCL_OP_REDUCE_MAP[op];
2393 7 : opParams.sendBuf = sendBuf;
2394 7 : opParams.recvBuf = recvBuf;
2395 7 : opParams.count = recvCount;
2396 7 : opParams.opTag = tag;
2397 :
2398 7 : CHK_RET_AND_PRINT_IDE(communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream)), tag.c_str());
2399 :
2400 7 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2401 3 : HcclUs endut = TIME_NOW();
2402 : /* 关键状态记录 */
2403 : std::string endInfo
2404 6 : = "HcclReduceScatterV2:success,take time: " + std::to_string(DURATION_US(endut - startut).count())
2405 9 : + " us, tag: " + tag + std::string(stackLogBufferV2);
2406 3 : communicator->GetTrace().Save(endInfo);
2407 3 : }
2408 :
2409 7 : return HCCL_SUCCESS;
2410 9 : }
2411 :
2412 10 : HcclResult HcclReduceScatterVV2(
2413 : void* sendBuf, void* sendCounts, void* sendDispls, void* recvBuf, uint64_t recvCount, HcclDataType dataType,
2414 : HcclReduceOp op, HcclComm comm, aclrtStream stream)
2415 : {
2416 10 : HcclUs startut = TIME_NOW();
2417 : bool isCapture;
2418 10 : rtModel_t rtModel = nullptr;
2419 10 : u32 modelId = 0;
2420 :
2421 : // 获取通信域
2422 10 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
2423 10 : const std::string tag = "ReduceScatterV_" + communicator->GetId();
2424 :
2425 : // 获取rank信息
2426 : uint32_t rankId;
2427 10 : CHK_RET(communicator->GetRankId(rankId));
2428 : uint32_t rankSize;
2429 10 : CHK_RET(communicator->GetRankSize(&rankSize));
2430 : // 参数合法性校验
2431 10 : if (rankSize == 1) {
2432 : // rankSize为1时，退化为ReduceScatter
2433 : // 检查异常回退RSV的情况
2434 3 : if (recvCount == 0) {
2435 3 : HCCL_WARNING("[ReduceScatterV] recvCount is 0 when single rank");
2436 1 : return HCCL_SUCCESS;
2437 : } else {
2438 5 : CHK_PRT_RET(
2439 : sendBuf == nullptr, HCCL_ERROR("[ReduceScatterV] sendBuf is null when single rank"), HCCL_E_PTR);
2440 1 : CHK_PRT_RET(
2441 : recvBuf == nullptr, HCCL_ERROR("[ReduceScatterV] recvBuf is null when single rank"), HCCL_E_PTR);
2442 : }
2443 1 : return HcclReduceScatterV2(sendBuf, recvBuf, recvCount, dataType, op, comm, stream);
2444 : }
2445 13 : CHK_RET_AND_PRINT_IDE(HcomCheckOpParamV2(tag.c_str(), recvCount, dataType, stream), tag.c_str());
2446 6 : CHK_RET_AND_PRINT_IDE(HcomCheckReductionOpV2(op), tag.c_str());
2447 12 : CHK_RET_AND_PRINT_IDE(HcomCheckReduceDataTypeV2(dataType, op), tag.c_str());
2448 17 : CHK_RET_AND_PRINT_IDE(HcomCheckVOpParamV2(rankId, rankSize, recvCount, sendCounts), tag.c_str());
2449 3 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
2450 :
2451 : /* 接口交互信息日志 */
2452 : char stackLogBufferV2[LOG_TMPBUF_SIZE];
2453 3 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2454 0 : s32 streamId = HrtGetStreamId(stream);
2455 0 : s32 deviceLogicId = HrtGetDevice();
2456 0 : u32 localRank = INVALID_VALUE_RANKID;
2457 0 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
2458 :
2459 0 : s32 ret = snprintf_s(
2460 : stackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
2461 : "tag[%s], sendBuf[%p], recvBuf[%p], sendCounts[%p], sendDispls[%p], recvCount[%llu], dataType[%s], op[%s],"
2462 : "localRank[%u], streamId[%d], deviceLogicId[%d]",
2463 0 : tag.c_str(), sendBuf, recvBuf, sendCounts, sendDispls, recvCount, GetDataTypeEnumStrV2(dataType).c_str(),
2464 0 : GetReduceOpEnumStrV2(op).c_str(), localRank, streamId, deviceLogicId);
2465 :
2466 0 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
2467 0 : std::string logInfo = "Entry-HcclReduceScatterVV2:" + std::string(stackLogBufferV2);
2468 0 : if (isCapture) {
2469 0 : CHK_PTR_NULL(rtModel);
2470 : // 获取不到modelId会报错
2471 0 : CHK_RET(GetModelId(rtModel, modelId));
2472 0 : logInfo += ", model id[" + to_string(modelId) + "].";
2473 : }
2474 0 : communicator->GetTrace().Save(logInfo);
2475 0 : }
2476 :
2477 3 : u64* counts = static_cast<u64*>(sendCounts);
2478 3 : u64 inputCount = 0;
2479 9 : for (size_t index = 0; index < rankSize; index++) {
2480 6 : inputCount += counts[index];
2481 : }
2482 3 : if (inputCount == 0) {
2483 3 : HCCL_INFO("[%s] inputCount[%llu] is equal to zero", __func__, inputCount);
2484 1 : return HCCL_SUCCESS;
2485 : }
2486 2 : RPT_INPUT_ERR(
2487 : sendBuf == nullptr, "EI0003", std::vector<std::string>({"ccl_op", "value", "parameter", "expect"}),
2488 : std::vector<std::string>({"HcclReduceScatterVV2", "nullptr", "sendBuf", "not nullptr"}));
2489 2 : CHK_PTR_NULL(sendBuf);
2490 2 : if (op == HCCL_REDUCE_PROD) {
2491 0 : HCCL_ERROR("[Check][ReductionOp] Op:[HCCL_REDUCE_PROD] not supported, tag[%s]", tag.c_str());
2492 0 : return HCCL_E_NOT_SUPPORT;
2493 : }
2494 : // opParams组装
2495 2 : Hccl::CollOpParams opParams;
2496 2 : opParams.opType = Hccl::OpType::REDUCESCATTERV;
2497 2 : opParams.dataType = HcclDataTypeToDataType(dataType);
2498 2 : opParams.reduceOp = HcclReduceOpToReduceOp(op);
2499 2 : opParams.dstRank = rankId;
2500 2 : opParams.sendBuf = sendBuf;
2501 2 : opParams.recvBuf = recvBuf;
2502 2 : opParams.count = recvCount;
2503 2 : opParams.vDataDes.counts = sendCounts;
2504 2 : opParams.vDataDes.displs = sendDispls;
2505 2 : opParams.vDataDes.dataType = HcclDataTypeToDataType(dataType);
2506 :
2507 2 : CHK_RET_AND_PRINT_IDE(communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream)), tag.c_str());
2508 :
2509 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2510 0 : HcclUs endut = TIME_NOW();
2511 : /* 关键状态记录 */
2512 : std::string endInfo
2513 0 : = "HcclReduceScatterVV2:success,take time: " + std::to_string(DURATION_US(endut - startut).count())
2514 0 : + " us, tag: " + tag + std::string(stackLogBufferV2);
2515 0 : communicator->GetTrace().Save(endInfo);
2516 0 : }
2517 :
2518 2 : return HCCL_SUCCESS;
2519 10 : }
2520 :
2521 2 : HcclResult HcclBatchSendRecvV2(HcclSendRecvItem* sendRecvInfo, uint32_t itemNum, HcclComm comm, aclrtStream stream)
2522 : {
2523 2 : HcclUs startut = TIME_NOW();
2524 : bool isCapture;
2525 2 : rtModel_t rtModel = nullptr;
2526 2 : u32 modelId = 0;
2527 :
2528 2 : CHK_PTR_NULL(comm);
2529 2 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
2530 2 : const std::string tag = "HcclBatchSendRecvV2_" + communicator->GetId();
2531 2 : CHK_PTR_NULL(stream);
2532 2 : CHK_PTR_NULL(sendRecvInfo);
2533 2 : CHK_PRT_RET(itemNum == 0, HCCL_WARNING("[BatchSendRecv] taskList itemNum is zero."), HCCL_SUCCESS);
2534 2 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
2535 2 : u32 rankSize = 0;
2536 2 : CHK_RET(communicator->GetRankSize(&rankSize)); // 获取rankSize, 后续需要校验userrank是否在0到rankSize-1之间
2537 22 : for (uint32_t i = 0; i < itemNum; i++) {
2538 20 : if ((sendRecvInfo + i)->buf == nullptr) {
2539 8 : continue; // 支持数据量为0的场景，buf为空的跳过
2540 : }
2541 12 : CHK_RET_AND_PRINT_IDE(
2542 : HcomCheckOpParamV2(tag.c_str(), (sendRecvInfo + i)->count, (sendRecvInfo + i)->dataType, stream),
2543 : tag.c_str());
2544 12 : CHK_RET(HcomCheckUserRankV2(rankSize, (sendRecvInfo + i)->remoteRank));
2545 : }
2546 :
2547 : /* 记录接口交互信息日志 */
2548 : char stackLogBufferV2[LOG_TMPBUF_SIZE];
2549 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2550 1 : s32 streamId = HrtGetStreamId(stream);
2551 1 : s32 deviceLogicId = HrtGetDevice();
2552 1 : u32 localRank = INVALID_VALUE_RANKID;
2553 1 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
2554 :
2555 1 : s32 ret = snprintf_s(
2556 : stackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
2557 : "tag[%s], itemNum[%u], localRank[%u], streamId[%d], deviceLogicId[%d]", tag.c_str(), itemNum, localRank,
2558 : streamId, deviceLogicId);
2559 :
2560 1 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
2561 1 : std::string logInfo = "Entry-HcclBatchSendRecvV2:" + std::string(stackLogBufferV2);
2562 1 : if (isCapture) {
2563 0 : CHK_PTR_NULL(rtModel);
2564 0 : CHK_RET(GetModelId(rtModel, modelId));
2565 0 : logInfo += ", model id[" + to_string(modelId) + "].";
2566 : }
2567 1 : communicator->GetTrace().Save(logInfo);
2568 1 : }
2569 :
2570 2 : static thread_local Hccl::CollOpParams opParams;
2571 2 : opParams.opType = Hccl::OpType::BATCHSENDRECV;
2572 2 : opParams.batchSendRecvDataDes.sendRecvItemsPtr = static_cast<void*>(sendRecvInfo);
2573 2 : opParams.batchSendRecvDataDes.itemNum = itemNum;
2574 2 : opParams.dataType = HcclDataTypeToDataType(sendRecvInfo->dataType);
2575 :
2576 2 : CHK_RET_AND_PRINT_IDE(communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream)), tag.c_str());
2577 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2578 1 : HcclUs endut = TIME_NOW();
2579 : std::string endInfo
2580 2 : = "HcclBatchSendRecvV2:success,take time: " + std::to_string(DURATION_US(endut - startut).count())
2581 3 : + " us, tag: " + tag + std::string(stackLogBufferV2);
2582 1 : communicator->GetTrace().Save(endInfo);
2583 1 : }
2584 2 : return HCCL_SUCCESS;
2585 2 : }
2586 :
2587 : // 功能说明：推动式建链，超时退出
2588 2 : HcclResult WaitAllCommReady(s32 deviceLogicId)
2589 : {
2590 : try {
2591 2 : HrtSetDevice(deviceLogicId);
2592 2 : HcclCommInfoV2& opbasedCommInfoV2 = GetCommInfoV2();
2593 : // 定义最大等待10秒
2594 2 : constexpr u32 waitTransportReadyTimeoutMs = 10 * 1000; // 待解决
2595 2 : auto timeout = std::chrono::milliseconds(waitTransportReadyTimeoutMs);
2596 2 : HcclUs startTime = std::chrono::steady_clock::now();
2597 :
2598 : // 创建锁，防止在建链过程中，依旧还在恢复通信域
2599 2 : std::unique_lock<std::mutex> groupParaLock(opbasedCommInfoV2.groupParamsLock);
2600 6 : HCCL_INFO("[%s] deviceLogicId[%d] start wait all comm ready", __func__, deviceLogicId);
2601 : // 轮巡调度，推动式建链
2602 : while (true) {
2603 2 : bool isAllCommReady = true;
2604 : Hccl::HcclCommunicator* communicator;
2605 : // 枚举所有通信域进行推动式建链
2606 5 : for (auto iter = opbasedCommInfoV2.hcclGroupMap.begin(); iter != opbasedCommInfoV2.hcclGroupMap.end();
2607 3 : iter++) {
2608 3 : CHK_PTR_NULL(iter->second.pComm);
2609 3 : communicator = static_cast<Hccl::HcclCommunicator*>(iter->second.pComm.get());
2610 3 : if (!communicator->IsCommReady()) {
2611 : // 只要任意Comm一个没有ready，整体建链结果为 false
2612 1 : isAllCommReady = false;
2613 : }
2614 : }
2615 2 : if (isAllCommReady) {
2616 1 : break;
2617 : }
2618 :
2619 : // 超时判断
2620 1 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
2621 3 : HCCL_ERROR("[%s]WaitAllCommReady timeout, deviceLogicId[%d].", __func__, deviceLogicId);
2622 1 : return HCCL_E_TIMEOUT;
2623 : }
2624 0 : }
2625 1 : HrtResetDevice(deviceLogicId);
2626 2 : } catch (HcclException& e) {
2627 0 : HCCL_ERROR(e.what());
2628 0 : return e.GetErrorCode();
2629 0 : } catch (std::exception& e) {
2630 0 : HCCL_ERROR(e.what());
2631 0 : return HCCL_E_INTERNAL;
2632 0 : } catch (...) {
2633 0 : HCCL_ERROR("Unknown error occurs!");
2634 0 : return HCCL_E_INTERNAL;
2635 0 : }
2636 :
2637 3 : HCCL_INFO("[%s] all comm ready.", __func__);
2638 1 : return HCCL_SUCCESS;
2639 : }
2640 :
2641 : #ifdef __cplusplus
2642 : extern "C" {
2643 : #endif // __cplusplus
2644 :
2645 : HcclResult
2646 0 : HcclGetCclBuffer(HcclComm comm, uintptr_t& cclBufferAddr, size_t& cclBufferSize, HcclMemType& cclBufferMemType)
2647 : {
2648 0 : Hccl::HcclCommunicator* communicatorV2 = (static_cast<Hccl::HcclCommunicator*>(comm));
2649 0 : CHK_PTR_NULL(communicatorV2);
2650 0 : CHK_RET(communicatorV2->HcclGetCclBuffer(cclBufferAddr, cclBufferSize, cclBufferMemType));
2651 0 : return HCCL_SUCCESS;
2652 : }
2653 :
2654 1 : HcclResult HcclGetRawCommHandle(const char* commName, HcclComm* commHandle)
2655 : {
2656 1 : CHK_PTR_NULL(commName);
2657 1 : CHK_PTR_NULL(commHandle);
2658 :
2659 1 : HcclCommInfoV2& opbasedCommInfoV2 = GetCommInfoV2();
2660 1 : std::unique_lock<std::mutex> lock(opbasedCommInfoV2.groupParamsLock);
2661 3 : HCCL_INFO("[HcclGetRawCommHandle] group:[%s]", commName);
2662 2 : auto iter = opbasedCommInfoV2.hcclGroupMap.find(commName);
2663 1 : if (iter == opbasedCommInfoV2.hcclGroupMap.end()) {
2664 0 : HCCL_ERROR("[HcclGetRawCommHandle] commName [%s] not found, please check.", commName);
2665 0 : return HCCL_E_PARA;
2666 : }
2667 2 : *commHandle = static_cast<HcclComm>(opbasedCommInfoV2.hcclGroupMap[commName].pComm.get());
2668 1 : return HCCL_SUCCESS;
2669 1 : }
2670 :
2671 2 : HcclResult HcclGetCcuTaskInfoLegacy(HcclComm comm, void* tilingData, void* ccuTaskGroup)
2672 : {
2673 2 : CHK_PTR_NULL(comm);
2674 2 : CHK_PTR_NULL(tilingData);
2675 2 : CHK_PTR_NULL(ccuTaskGroup);
2676 :
2677 2 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
2678 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2679 0 : HCCL_RUN_INFO(
2680 : "Entry-HcclGetCcuTaskInfo V950, commId[%s], tilingData[%p], ccuTaskGroup[%p]",
2681 : communicator->GetId().c_str(), tilingData, ccuTaskGroup);
2682 : }
2683 2 : auto ret = communicator->GetCcuTaskInfo(tilingData, ccuTaskGroup);
2684 2 : if (ret != HCCL_SUCCESS) {
2685 3 : HCCL_ERROR("HcclGetCcuTaskInfo ret[%d] commId[%s]", ret, communicator->GetId().c_str());
2686 1 : return HCCL_E_INTERNAL;
2687 : }
2688 :
2689 1 : return HCCL_SUCCESS;
2690 : }
2691 :
2692 2 : HcclResult HcclSnapshotSave(void* snapshotBuf, uint32_t size, uint32_t step)
2693 : {
2694 : // 快照保存只支持ranktable场景，不支持topo探测场景
2695 6 : HCCL_INFO("[%s] snapshot save start, size[%u], step[%u]", __func__, size, step);
2696 2 : CHK_PTR_NULL(snapshotBuf);
2697 : // 获取公共信息
2698 2 : HcclCommInfoV2& opbasedCommInfoV2 = GetCommInfoV2();
2699 : // 获取公共信息里的step信息
2700 2 : u64 savedStep = opbasedCommInfoV2.step;
2701 : // 如果当前step不等于用户传入的step，直接返回报错
2702 2 : if (savedStep != step) {
2703 0 : HCCL_ERROR("[%s] step is not match, savedStep[%u], userInputStep[%u]", __func__, savedStep, step);
2704 0 : return HCCL_E_PARA;
2705 : }
2706 : // 获取保存的快照
2707 2 : Hccl::BinaryStream& savedSnapshotBuf = Hccl::SnapShotParser::GetInstance().GetSnapShotBuf();
2708 2 : uint32_t dataLen = savedSnapshotBuf.GetSize();
2709 6 : HCCL_INFO("[%s] savedSnapshotBuf data len[%u]", __func__, dataLen);
2710 :
2711 : // 获取保存的size
2712 2 : if ((dataLen + sizeof(dataLen) + sizeof(uint32_t)) != size) {
2713 : // 如果当前size不等于用户传入的size，直接返回报错
2714 3 : HCCL_ERROR("[%s] size is not match, userInputSize[%u] dataLen[%u]", __func__, size, dataLen);
2715 1 : return HCCL_E_PARA;
2716 : }
2717 1 : std::vector<char> data;
2718 1 : savedSnapshotBuf.DumpWithRevert(data);
2719 3 : HCCL_INFO("[%s] dump data size [%u]", __func__, data.size());
2720 1 : if (data.empty()) {
2721 3 : HCCL_INFO("[%s] dump data empty", __func__);
2722 1 : return HCCL_E_INTERNAL;
2723 : }
2724 :
2725 : // 计算快照crc值
2726 0 : uint32_t crcValue{0};
2727 0 : CHK_RET(Hccl::SnapShotParser::GetInstance().CalcBufCrc32(savedSnapshotBuf, crcValue));
2728 :
2729 : // 将快照大小拷贝到用户传入的内存中
2730 0 : s32 sRet = memcpy_s(snapshotBuf, size, static_cast<void*>(&dataLen), sizeof(dataLen));
2731 0 : CHK_PRT_RET(sRet != EOK, HCCL_ERROR("[%s] memcpy dataLen failed, return[%d]", __func__, sRet), HCCL_E_MEMORY);
2732 : // 将快照crc值拷贝到用户传入的内存中
2733 0 : sRet = memcpy_s(
2734 0 : static_cast<char*>(snapshotBuf) + sizeof(dataLen), size - sizeof(dataLen), static_cast<void*>(&crcValue),
2735 : sizeof(crcValue));
2736 0 : CHK_PRT_RET(sRet != EOK, HCCL_ERROR("[%s] memcpy crcValue failed, return[%d]", __func__, sRet), HCCL_E_MEMORY);
2737 : // 将快照拷贝到用户传入的内存中
2738 0 : sRet = memcpy_s(static_cast<char*>(snapshotBuf) + sizeof(dataLen) + sizeof(crcValue), dataLen, &data[0], dataLen);
2739 0 : CHK_PRT_RET(sRet != EOK, HCCL_ERROR("[%s] memcpy data failed, return[%d]", __func__, sRet), HCCL_E_MEMORY);
2740 0 : return HCCL_SUCCESS;
2741 1 : }
2742 :
2743 1 : void RecoverSnapshotCcuStatus(const std::shared_ptr<Hccl::SnapShotBuf>& savedSnapshotBuf)
2744 : {
2745 1 : HcclCommInfoV2& opbasedCommInfoV2 = GetCommInfoV2();
2746 : // 通过进程锁看护，避免多个通信域同时占用CCU_MS
2747 1 : std::unique_lock<std::mutex> lock(opbasedCommInfoV2.groupParamsLock);
2748 :
2749 1 : for (auto useMsCommId : savedSnapshotBuf->ccuStatusSnapshot.useMsCommIds) {
2750 0 : opbasedCommInfoV2.ccuStatus.useMsCommIds.push_back(useMsCommId.data());
2751 : }
2752 :
2753 1 : for (auto useSchedCommId : savedSnapshotBuf->ccuStatusSnapshot.useSchedCommIds) {
2754 0 : opbasedCommInfoV2.ccuStatus.useSchedCommIds.push_back(useSchedCommId.data());
2755 : }
2756 : // 结果打印
2757 3 : HCCL_INFO(
2758 : "RecoverSnapshotCcuStatus ccuMsCommIds size[%lu] ccuSchedCommIds size[%lu]",
2759 : opbasedCommInfoV2.ccuStatus.useMsCommIds.size(), opbasedCommInfoV2.ccuStatus.useSchedCommIds.size());
2760 1 : for (auto useMsCommId : opbasedCommInfoV2.ccuStatus.useMsCommIds) {
2761 0 : HCCL_DEBUG("RecoverSnapshotCcuStatus useMsCommId[%s]", useMsCommId.c_str());
2762 0 : }
2763 1 : for (auto useSchedCommId : opbasedCommInfoV2.ccuStatus.useSchedCommIds) {
2764 0 : HCCL_DEBUG("RecoverSnapshotCcuStatus useSchedCommId[%s]", useSchedCommId.c_str());
2765 0 : }
2766 1 : }
2767 :
2768 1 : HcclResult HcclSnapshotRecoverAllComms(
2769 : const char* clusterInfo, const char* changedInfo, void* snapshotBuf, uint32_t snapshotBufSize)
2770 : {
2771 : (void)clusterInfo;
2772 : // 入参校验，clusterInfo和changedInfo暂时用不到，先不判空
2773 3 : HCCL_INFO("[%s] snapshot recover start.", __func__);
2774 1 : CHK_PTR_NULL(snapshotBuf);
2775 :
2776 : // 当前不支持在不下发算子的情况下重复调用recover接口
2777 1 : if (Hccl::SnapShotParser::GetInstance().GetIsNeedLoadOp()) {
2778 0 : HCCL_ERROR("[%s] snapshot recover failed, it is necessary to use it after load op, please check!", __func__);
2779 0 : return HCCL_E_INTERNAL;
2780 : }
2781 :
2782 1 : s32 deviceLogicId = HcclGetThreadDeviceId();
2783 :
2784 1 : CHK_RET(CallSingletons()); // 临时规避，在初始化通信域前声明单例保证时序
2785 :
2786 : // 获取公共信息
2787 1 : HcclCommInfoV2& opbasedCommInfoV2 = GetCommInfoV2();
2788 : // 调用恢复快照公共信息函数，让字节流的struct数据恢复到savedSnapshotBuf
2789 1 : std::shared_ptr<Hccl::SnapShotBuf> savedSnapshotBuf = std::make_shared<Hccl::SnapShotBuf>();
2790 1 : CHK_RET(
2791 : Hccl::SnapShotParser::GetInstance().ParseSnapshotToLocalBuff(snapshotBuf, snapshotBufSize, *savedSnapshotBuf));
2792 :
2793 1 : opbasedCommInfoV2.step = savedSnapshotBuf->snapShotPub.step;
2794 : // 将状态设置为RECOVERED
2795 1 : opbasedCommInfoV2.status = DeviceStatus::DEVICE_RECOVERED;
2796 :
2797 : // 创建通信域
2798 1 : opbasedCommInfoV2.pComm.reset(new (std::nothrow) Hccl::HcclCommunicator(
2799 1 : savedSnapshotBuf->snapshot.snapShotComm.commParams, &savedSnapshotBuf->snapshot.snapShotComm.config));
2800 1 : CHK_PTR_NULL(opbasedCommInfoV2.pComm);
2801 1 : opbasedCommInfoV2.pComm->RegisterAcceStateCallBack(CommunicatorCallback());
2802 : // 恢复全局通讯域
2803 3 : HCCL_INFO("[%s] recover global group.", __func__);
2804 1 : CHK_RET(opbasedCommInfoV2.pComm->RecoverComm(
2805 : static_cast<void*>(&savedSnapshotBuf->snapshot.snapShotComm), savedSnapshotBuf->snapShotPub.step, changedInfo));
2806 3 : HCCL_INFO("[%s] global group recover success.", __func__);
2807 1 : opbasedCommInfoV2.commParams = savedSnapshotBuf->snapshot.snapShotComm.commParams;
2808 :
2809 1 : HcclGroupParamsV2 params{};
2810 1 : params.pComm = opbasedCommInfoV2.pComm;
2811 2 : opbasedCommInfoV2.hcclGroupMap[savedSnapshotBuf->snapshot.groupName] = params;
2812 :
2813 : // 这里变成多线程后，封装成小函数
2814 2 : for (uint32_t index = 0; index < savedSnapshotBuf->groupNum; index++) {
2815 1 : std::string groupName(savedSnapshotBuf->subSnapshot[index].groupName);
2816 : // 别重复恢复全局通讯域
2817 1 : if (groupName == savedSnapshotBuf->snapshot.groupName) {
2818 0 : continue;
2819 : }
2820 1 : const Hccl::SnapShotSubComm* snapShotSubComm = &savedSnapshotBuf->subSnapshot[index].snapShotSubComm;
2821 : // 创建一个指针
2822 : std::shared_ptr<Hccl::HcclCommunicator> commImp
2823 1 : = make_shared<Hccl::HcclCommunicator>(snapShotSubComm->commParams, &snapShotSubComm->config);
2824 : // 循环创建恢复子通讯域
2825 3 : HCCL_INFO("[%s] recover sub group[%s].", __func__, groupName.c_str());
2826 1 : CHK_RET(opbasedCommInfoV2.pComm->RecoverSubComm(
2827 : static_cast<const void*>(snapShotSubComm), commImp, savedSnapshotBuf->snapShotPub.step));
2828 3 : HCCL_INFO("[%s] sub group[%s] recover success.", __func__, groupName.c_str());
2829 1 : HcclGroupParamsV2 params{};
2830 1 : params.pComm = commImp;
2831 : // 这里进行多线程后，加锁
2832 1 : opbasedCommInfoV2.hcclGroupMap[groupName] = params;
2833 1 : }
2834 1 : RecoverSnapshotCcuStatus(savedSnapshotBuf);
2835 3 : HCCL_INFO("[%s] snapshot recover success.", __func__);
2836 1 : std::unique_ptr<std::thread> waitReadyThread;
2837 1 : waitReadyThread.reset(new (std::nothrow) std::thread(&WaitAllCommReady, deviceLogicId));
2838 1 : if (waitReadyThread == nullptr) {
2839 0 : HCCL_ERROR("[%s] new waitReadyThread failed.", __func__);
2840 0 : return HCCL_E_INTERNAL;
2841 : }
2842 : // 异步方案目前未分析清楚，采用临时方案规避
2843 1 : waitReadyThread->join();
2844 1 : Hccl::SnapShotParser::GetInstance().SetIsNeedLoadOp(true);
2845 3 : HCCL_INFO("[%s] all comm ready.", __func__);
2846 1 : return HCCL_SUCCESS;
2847 1 : }
2848 3 : static HcclResult GetAllSnapShotStaticBuf(
2849 : const std::shared_ptr<Hccl::HcclCommunicator>& pComm,
2850 : const std::map<std::string, std::shared_ptr<Hccl::HcclCommunicator>>& hcclGroupMap, uint32_t step,
2851 : Hccl::BinaryStream& buf)
2852 : {
2853 9 : HCCL_INFO("[%s] start", __func__);
2854 3 : if (!pComm->IsWorldGroup()) {
2855 0 : HCCL_ERROR("[%s] input comm is not hccl_world_group, please check!", __func__);
2856 0 : return HCCL_E_INTERNAL;
2857 : }
2858 : // 生成 静态流公共数据,直接往buf写
2859 3 : Hccl::SnapShotParser::GetInstance().SerializeCommVersionInfo(buf);
2860 : // 存储 全局通信域名称
2861 3 : buf << pComm->GetId();
2862 : // 存储 全局通信域 静态buf
2863 3 : Hccl::BinaryStream& pCommBuf = *(static_cast<Hccl::BinaryStream*>(pComm->GetStaticBinaryInfo()));
2864 3 : std::vector<char> pCommChars{};
2865 3 : pCommBuf.Dump(pCommChars);
2866 3 : for (auto c : pCommChars) {
2867 0 : buf << c;
2868 : }
2869 :
2870 3 : size_t pSubCommSize = hcclGroupMap.size() - 1;
2871 9 : HCCL_INFO("[%s] pSubCommSize[%u]", __func__, pSubCommSize);
2872 3 : buf << pSubCommSize; // 子通信域 静态buf 数量
2873 :
2874 3 : auto iter = hcclGroupMap.begin();
2875 6 : for (; iter != hcclGroupMap.end(); iter++) {
2876 3 : auto pSubCommName = iter->first;
2877 3 : if (pComm->GetId() == pSubCommName) {
2878 2 : continue;
2879 : }
2880 1 : buf << pSubCommName; // 子通信域 名字
2881 : // 获取子通信域 静态buf
2882 1 : auto groupComm = iter->second;
2883 1 : Hccl::BinaryStream& pSubCommBuf = *(static_cast<Hccl::BinaryStream*>(groupComm->GetStaticBinaryInfo()));
2884 1 : std::vector<char> pSubCommChars{};
2885 1 : pSubCommBuf.Dump(pSubCommChars);
2886 1 : for (auto c : pSubCommChars) {
2887 0 : buf << c;
2888 : }
2889 3 : }
2890 3 : buf << step;
2891 9 : HCCL_INFO("[%s] end", __func__);
2892 :
2893 3 : return HCCL_SUCCESS;
2894 3 : }
2895 :
2896 3 : static HcclResult GetAllSnapShotDynamicBuf(
2897 : const std::shared_ptr<Hccl::HcclCommunicator>& pComm,
2898 : const std::map<std::string, std::shared_ptr<Hccl::HcclCommunicator>>& hcclGroupMap, Hccl::BinaryStream& buf)
2899 : {
2900 9 : HCCL_INFO("[%s] start", __func__);
2901 3 : CHK_PTR_NULL(pComm);
2902 : // 全局通信域 动态buf信息
2903 3 : CHK_RET(pComm->GetSnapShotDynamicBuf(static_cast<void*>(&buf)));
2904 :
2905 3 : size_t pSubCommSize = hcclGroupMap.size() - 1;
2906 3 : buf << pSubCommSize; // 子通信域 动态buf信息 数量
2907 9 : HCCL_INFO("[%s] hcclGroupMap size[%u]", __func__, hcclGroupMap.size());
2908 3 : auto iter = hcclGroupMap.begin();
2909 6 : for (; iter != hcclGroupMap.end(); iter++) {
2910 3 : auto pSubCommName = iter->first;
2911 3 : if (pComm->GetId() == pSubCommName) {
2912 2 : continue;
2913 : }
2914 : // 获取子通信域 建链邻居信息buf
2915 1 : CHK_RET(iter->second->GetSnapShotDynamicBuf(static_cast<void*>(&buf)));
2916 3 : }
2917 9 : HCCL_INFO("[%s] end", __func__);
2918 3 : return HCCL_SUCCESS;
2919 : }
2920 :
2921 3 : void GetSnapShotCcuStatusBuf(Hccl::BinaryStream& buf)
2922 : {
2923 3 : HcclCommInfoV2& opbasedCommInfoV2 = GetCommInfoV2();
2924 : // 通过进程锁看护，避免多个通信域同时占用CCU_MS
2925 3 : std::unique_lock<std::mutex> lock(opbasedCommInfoV2.groupParamsLock);
2926 :
2927 3 : auto ccuStatus = opbasedCommInfoV2.ccuStatus;
2928 3 : buf << ccuStatus.useMsCommIds.size();
2929 9 : HCCL_INFO("useMsCommIds size is %u", ccuStatus.useMsCommIds.size());
2930 3 : for (auto useMsCommId : ccuStatus.useMsCommIds) {
2931 0 : buf << useMsCommId;
2932 0 : HCCL_INFO("useMsCommId is %s", useMsCommId.c_str());
2933 0 : }
2934 :
2935 3 : buf << ccuStatus.useSchedCommIds.size();
2936 9 : HCCL_INFO("useSchedCommIds size is %u", ccuStatus.useSchedCommIds.size());
2937 3 : for (auto useSchedCommId : ccuStatus.useSchedCommIds) {
2938 0 : buf << useSchedCommId;
2939 0 : HCCL_INFO("useSchedCommId is %s", useSchedCommId.c_str());
2940 0 : }
2941 3 : }
2942 :
2943 : // 获取快照占用buffer的大小 --给op_base API getSize调，内部生成完整长流。save后释放
2944 3 : HcclResult SnapshotGenerate(
2945 : const std::shared_ptr<Hccl::HcclCommunicator>& pComm,
2946 : const std::map<std::string, std::shared_ptr<Hccl::HcclCommunicator>>& hcclGroupMap, uint32_t step, uint32_t* size)
2947 : {
2948 3 : CHK_PTR_NULL(size);
2949 3 : CHK_PTR_NULL(pComm);
2950 9 : HCCL_INFO("[%s] start", __func__);
2951 3 : Hccl::SnapShotParser::GetInstance().GetSnapShotBuf().Clear();
2952 3 : CHK_RET(GetAllSnapShotStaticBuf(pComm, hcclGroupMap, step, Hccl::SnapShotParser::GetInstance().GetSnapShotBuf()));
2953 3 : CHK_RET(GetAllSnapShotDynamicBuf(pComm, hcclGroupMap, Hccl::SnapShotParser::GetInstance().GetSnapShotBuf()));
2954 3 : GetSnapShotCcuStatusBuf(Hccl::SnapShotParser::GetInstance().GetSnapShotBuf());
2955 : // size = 流长度 + crc（u32）长度 + 存储流长度所需长度
2956 3 : uint32_t dataLen = static_cast<uint32_t>(Hccl::SnapShotParser::GetInstance().GetSnapShotBuf().GetSize());
2957 3 : *size = dataLen + sizeof(dataLen) + sizeof(uint32_t); // 快照头上保存一个总长度和crc长度
2958 9 : HCCL_INFO("[%s] end, size[%u]", __func__, *size);
2959 3 : return HCCL_SUCCESS;
2960 : }
2961 :
2962 2 : HcclResult HcclSnapshotGetBufSize(uint32_t step, uint32_t* size)
2963 : {
2964 : // 校验DevType
2965 6 : HCCL_INFO("[%s] start", __func__);
2966 2 : Hccl::DevType devType = HrtGetDeviceType();
2967 2 : if (devType != DevType::DEV_TYPE_950 && devType != DevType::DEV_TYPE_960) {
2968 0 : HCCL_INFO("[%s] Get buffer size not support in this device type[%d]", __func__, devType);
2969 0 : return HCCL_E_NOT_SUPPORT;
2970 : }
2971 :
2972 : // step保存在g_opbasedCommInfoV2，save的时候校验，防止GetBufSize和Save的step不一致
2973 2 : HcclCommInfoV2& opbasedCommInfoV2 = GetCommInfoV2();
2974 2 : opbasedCommInfoV2.step = step;
2975 :
2976 2 : std::map<std::string, std::shared_ptr<Hccl::HcclCommunicator>> subCommMap;
2977 4 : for (auto hcclGroupMap : opbasedCommInfoV2.hcclGroupMap) {
2978 2 : subCommMap.insert(std::make_pair(hcclGroupMap.first, hcclGroupMap.second.pComm));
2979 2 : }
2980 2 : return SnapshotGenerate(opbasedCommInfoV2.pComm, subCommMap, step, size);
2981 2 : }
2982 :
2983 1 : HcclResult HcclGetTopoDescV2()
2984 : {
2985 3 : HCCL_ERROR("Current chip type does not support GetTopoDesc.");
2986 1 : return HCCL_E_NOT_SUPPORT;
2987 : }
2988 :
2989 1 : HcclResult HcclGetCommAsyncErrorV2() { return HCCL_SUCCESS; }
2990 :
2991 0 : HcclResult HcclSetConfigV2(HcclConfig config, HcclConfigValue configValue)
2992 : {
2993 : (void)(config);
2994 : (void)(configValue);
2995 0 : HCCL_WARNING("DETERMINISTIC_ENABLE is default option in 950! Can not set.");
2996 0 : return HCCL_SUCCESS;
2997 : }
2998 0 : HcclResult HcclGetConfigV2(HcclConfig config, HcclConfigValue* configValue)
2999 : {
3000 : (void)(config);
3001 0 : constexpr int32_t DETERMINISTIC_ENABLE = 1; // A5支持确定性，不需要配置
3002 0 : (*configValue).value = DETERMINISTIC_ENABLE;
3003 0 : HCCL_WARNING("DETERMINISTIC_ENABLE is default option in 950!");
3004 0 : return HCCL_SUCCESS;
3005 : }
3006 :
3007 1 : HcclResult HcclGetRankGraphV2(const HcclComm* comm, void** rankGraph)
3008 : {
3009 1 : Hccl::HcclCommunicator* communicatorV2 = (static_cast<Hccl::HcclCommunicator*>(*comm));
3010 1 : communicatorV2->GetRankGraphV2(*rankGraph);
3011 1 : return HCCL_SUCCESS;
3012 : }
3013 :
3014 0 : HcclResult HcommFlushV2()
3015 : {
3016 0 : HCCL_INFO("[HcommFlushV2]");
3017 0 : return FlushManager::GetInstance().Flush();
3018 : }
3019 :
3020 0 : HcclResult CommGetCCLBufSizeCfgV2(HcclComm comm, uint64_t* cclBufSize)
3021 : {
3022 0 : HCCL_RUN_INFO("Entry-CommGetCCLBufSizeCfg V950");
3023 0 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
3024 0 : CHK_RET(communicator->GetConfigInCCLbufferSize(cclBufSize));
3025 0 : return HCCL_SUCCESS;
3026 : }
3027 :
3028 0 : HcclResult HcclGetNetLayersV2(HcclComm comm, uint32_t** netLayers, uint32_t* netLayerNum)
3029 : {
3030 0 : HCCL_RUN_INFO("Entry-HcclGetNetLayersV2 V950");
3031 0 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
3032 0 : auto ret = communicator->GetNetLayers(netLayers, netLayerNum);
3033 0 : if (ret != HCCL_SUCCESS) {
3034 0 : HCCL_ERROR(
3035 : "HcclGetNetLayersV2 get netLayers from communicator failed, commId[%s], ret[%d]",
3036 : communicator->GetId().c_str(), ret);
3037 0 : return HCCL_E_NOT_FOUND; // 查询失败返回
3038 : }
3039 0 : return HCCL_SUCCESS;
3040 : }
3041 :
3042 0 : HcclResult HcclGetInstSizeByNetLayerV2(HcclComm comm, uint32_t netLayer, uint32_t* rankNum)
3043 : {
3044 0 : HCCL_RUN_INFO("Entry-HcclGetInstSizeByNetLayerV2 V950");
3045 0 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
3046 0 : auto ret = communicator->GetInstSizeByNetLayer(netLayer, rankNum);
3047 0 : if (ret != HCCL_SUCCESS) {
3048 0 : HCCL_ERROR(
3049 : "HcclGetInstSizeByNetLayerV2 get InstSize from communicator failed at netLayer[%u], commId[%s], ret[%d]",
3050 : netLayer, communicator->GetId().c_str(), ret);
3051 0 : return HCCL_E_NOT_FOUND; // 查询失败返回
3052 : }
3053 : /* 关键状态记录 */
3054 0 : HCCL_INFO(
3055 : "HcclGetInstSizeByNetLayerV2 success, netLayer[%u], rankNum[%u], commId[%s]", netLayer, *rankNum,
3056 : communicator->GetId().c_str());
3057 0 : return HCCL_SUCCESS;
3058 : }
3059 :
3060 2 : HcclResult HcclGetInstRanksByNetLayerV2(HcclComm comm, uint32_t netLayer, uint32_t** ranks, uint32_t* rankNum)
3061 : {
3062 6 : HCCL_RUN_INFO("Entry-HcclGetInstRanksByNetLayerV2 V950");
3063 2 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
3064 2 : auto ret = communicator->GetInstRanksByNetLayer(netLayer, ranks, rankNum);
3065 2 : if (ret != HCCL_SUCCESS) {
3066 3 : HCCL_ERROR(
3067 : "HcclGetInstRanksByNetLayerV2 get ranks from communicator failed at netLayer[%u], commId[%s], ret[%d]",
3068 : netLayer, communicator->GetId().c_str(), ret);
3069 1 : return HCCL_E_NOT_FOUND; // 查询失败返回
3070 : }
3071 : /* 关键状态记录 */
3072 3 : HCCL_INFO(
3073 : "HcclGetInstRanksByNetLayerV2 success, netLayer[%u], rankNum[%u], commId[%s]", netLayer, *rankNum,
3074 : communicator->GetId().c_str());
3075 1 : return HCCL_SUCCESS;
3076 : }
3077 :
3078 2 : HcclResult HcclGetInstTopoTypeByNetLayerV2(HcclComm comm, uint32_t netLayer, uint32_t* topoType)
3079 : {
3080 6 : HCCL_RUN_INFO("Entry-HcclGetInstTopoTypeByNetLayer V950");
3081 2 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
3082 2 : auto ret = communicator->GetInstTopoTypeByNetLayer(netLayer, topoType);
3083 2 : if (ret != HCCL_SUCCESS) {
3084 3 : HCCL_ERROR(
3085 : "HcclGetInstTopoTypeByNetLayerV2 get topoType from communicator failed at netLayer[%u], commId[%s], "
3086 : "ret[%d]",
3087 : netLayer, communicator->GetId().c_str(), ret);
3088 1 : return HCCL_E_NOT_FOUND;
3089 : }
3090 : /* 关键状态记录 */
3091 3 : HCCL_INFO(
3092 : "HcclGetInstTopoTypeByNetLayer success, netLayer[%u] topoType[%u] commId[%s]", netLayer, *topoType,
3093 : communicator->GetId().c_str());
3094 1 : return HCCL_SUCCESS;
3095 : }
3096 :
3097 : HcclResult
3098 2 : HcclGetInstSizeListByNetLayerV2(HcclComm comm, uint32_t netLayer, uint32_t** instSizeList, uint32_t* listSize)
3099 : {
3100 6 : HCCL_RUN_INFO("Entry-HcclGetInstSizeListByNetLayer V950");
3101 2 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
3102 2 : auto ret = communicator->GetInstSizeListByNetLayer(netLayer, instSizeList, listSize);
3103 2 : if (ret != HCCL_SUCCESS) {
3104 3 : HCCL_ERROR(
3105 : "HcclGetInstSizeListByNetLayerV2 get netInstance size from communicator failed at netLayer[%u], "
3106 : "commId[%s], ret[%d]",
3107 : netLayer, communicator->GetId().c_str(), ret);
3108 1 : return HCCL_E_NOT_FOUND;
3109 : }
3110 : /* 关键状态记录 */
3111 3 : HCCL_INFO(
3112 : "HcclGetInstSizeListByNetLayer success, netLayer[%u] listSize[%u] commId[%s]", netLayer, *listSize,
3113 : communicator->GetId().c_str());
3114 1 : return HCCL_SUCCESS;
3115 : }
3116 :
3117 2 : HcclResult HcclGetLinksV2(
3118 : HcclComm comm, uint32_t netLayer, uint32_t srcRank, uint32_t dstRank, CommLink** linkList, uint32_t* listSize)
3119 : {
3120 6 : HCCL_RUN_INFO("Entry-HcclGetInstSizeListByNetLayer V950");
3121 2 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
3122 2 : auto ret = communicator->GetLinks(netLayer, srcRank, dstRank, linkList, listSize);
3123 2 : if (ret != HCCL_SUCCESS) {
3124 3 : HCCL_ERROR(
3125 : "HcclGetLinksV2 get links from communicator failed at netLayer[%u], commId[%s], ret[%d]", netLayer,
3126 : communicator->GetId().c_str(), ret);
3127 1 : return HCCL_E_NOT_FOUND;
3128 : }
3129 : /* 关键状态记录 */
3130 3 : HCCL_INFO(
3131 : "HcclGetLinks success, netLayer[%u], srcRank[%u], dstRank[%u], listSize[%u], commId[%s]", netLayer, srcRank,
3132 : dstRank, *listSize, communicator->GetId().c_str());
3133 1 : return HCCL_SUCCESS;
3134 : }
3135 :
3136 2 : HcclResult HcclGetTopoInstsByLayerV2(HcclComm comm, uint32_t netLayer, uint32_t** topoInsts, uint32_t* topoInstNum)
3137 : {
3138 6 : HCCL_RUN_INFO("Entry-HcclGetTopoInstsByLayer V950");
3139 2 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
3140 2 : auto ret = communicator->GetTopoInstsByLayer(netLayer, topoInsts, topoInstNum);
3141 2 : if (ret != HCCL_SUCCESS) {
3142 3 : HCCL_ERROR(
3143 : "HcclGetTopoInstsByLayer get topoInsts from communicator failed at netLayer[%u], commId[%s], ret[%d]",
3144 : netLayer, communicator->GetId().c_str(), ret);
3145 1 : return HCCL_E_NOT_FOUND;
3146 : }
3147 : /* 关键状态记录 */
3148 3 : HCCL_INFO(
3149 : "HcclGetTopoInstsByLayer success, netLayer[%u], topoInstNum[%u], commId[%s]", netLayer, *topoInstNum,
3150 : communicator->GetId().c_str());
3151 1 : return HCCL_SUCCESS;
3152 : }
3153 :
3154 2 : HcclResult HcclGetTopoTypeV2(HcclComm comm, uint32_t netLayer, uint32_t topoInstId, CommTopo* topoType)
3155 : {
3156 6 : HCCL_RUN_INFO("Entry-HcclGetInstSizeListByNetLayer V950");
3157 2 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
3158 2 : auto ret = communicator->GetTopoType(netLayer, topoInstId, topoType);
3159 2 : if (ret != HCCL_SUCCESS) {
3160 3 : HCCL_ERROR(
3161 : "HcclGetTopoType get topoType from communicator failed at netLayer[%u], commId[%s], ret[%d]", netLayer,
3162 : communicator->GetId().c_str(), ret);
3163 1 : return HCCL_E_NOT_FOUND;
3164 : }
3165 : /* 关键状态记录 */
3166 3 : HCCL_INFO(
3167 : "HcclGetInstSizeListByNetLayer success, netLayer[%u] topoType[%u] commId[%s]", netLayer, *topoType,
3168 : communicator->GetId().c_str());
3169 1 : return HCCL_SUCCESS;
3170 : }
3171 :
3172 : HcclResult
3173 2 : HcclGetRanksByTopoInstV2(HcclComm comm, uint32_t netLayer, uint32_t topoInstId, uint32_t** ranks, uint32_t* rankNum)
3174 : {
3175 6 : HCCL_RUN_INFO("Entry-HcclGetRanksByTopoInst V950");
3176 2 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
3177 2 : auto ret = communicator->GetRanksByTopoInst(netLayer, topoInstId, ranks, rankNum);
3178 2 : if (ret != HCCL_SUCCESS) {
3179 3 : HCCL_ERROR(
3180 : "HcclGetTopoInstsByLayer get ranks from communicator failed at netLayer[%u], commId[%s], ret[%d]", netLayer,
3181 : communicator->GetId().c_str(), ret);
3182 1 : return HCCL_E_NOT_FOUND;
3183 : }
3184 : /* 关键状态记录 */
3185 3 : HCCL_INFO("HcclGetRanksByTopoInst success, netLayer[%u] rankNum[%u]", netLayer, *rankNum);
3186 1 : return HCCL_SUCCESS;
3187 : }
3188 :
3189 2 : HcclResult HcclRankGraphGetEndpointNumV2(HcclComm comm, uint32_t layer, uint32_t topoInstId, uint32_t* num)
3190 : {
3191 6 : HCCL_RUN_INFO("Entry-HcclRankGraphGetEndpointNum V950");
3192 2 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
3193 2 : auto ret = communicator->GetEndpointNum(layer, topoInstId, num);
3194 2 : if (ret != HCCL_SUCCESS) {
3195 3 : HCCL_ERROR(
3196 : "HcclRankGraphGetEndpointNum get endpoint num from communicator failed at netLayer[%u] with "
3197 : "topoInstId[%u], commId[%s], ret[%d]",
3198 : layer, topoInstId, communicator->GetId().c_str(), ret);
3199 1 : return HCCL_E_NOT_FOUND;
3200 : }
3201 1 : return HCCL_SUCCESS;
3202 : }
3203 :
3204 3 : HcclResult HcclRankGraphGetEndpointDescV2(
3205 : HcclComm comm, uint32_t layer, uint32_t topoInstId, uint32_t* descNum, EndpointDesc* endpointDesc)
3206 : {
3207 9 : HCCL_RUN_INFO("Entry-HcclRankGraphGetEndpointDesc V950");
3208 3 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
3209 3 : auto ret = communicator->GetEndpointDesc(layer, topoInstId, descNum, endpointDesc);
3210 3 : if (ret != HCCL_SUCCESS) {
3211 3 : HCCL_ERROR(
3212 : "HcclRankGraphGetEndpointDesc get endpoint desc from communicator failed at netLayer[%u], commId[%s], "
3213 : "ret[%d]",
3214 : layer, communicator->GetId().c_str(), ret);
3215 1 : return HCCL_E_NOT_FOUND;
3216 : }
3217 2 : return HCCL_SUCCESS;
3218 : }
3219 :
3220 1 : HcclResult HcclRankGraphGetEndpointInfoV2(
3221 : HcclComm comm, uint32_t rankId, const EndpointDesc* endpointDesc, EndpointAttr endpointAttr, uint32_t infoLen,
3222 : void* info)
3223 : {
3224 3 : HCCL_RUN_INFO("Entry-HcclRankGraphGetEndpointInfo V950");
3225 1 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
3226 1 : auto ret = communicator->GetEndpointInfo(rankId, endpointDesc, endpointAttr, infoLen, info);
3227 1 : if (ret != HCCL_SUCCESS) {
3228 0 : HCCL_ERROR(
3229 : "HcclRankGraphGetEndpointInfo get info from communicator failed with endpointAttr [%d], commId[%s], "
3230 : "ret[%d]",
3231 : static_cast<s32>(endpointAttr), communicator->GetId().c_str(), ret);
3232 0 : return HCCL_E_NOT_FOUND;
3233 : }
3234 1 : return HCCL_SUCCESS;
3235 : }
3236 :
3237 1 : HcclResult HcclCommWorkingDevNicSetV2(const HcclComm comm, uint32_t* ranks, bool* useBackup, uint32_t nRanks)
3238 : {
3239 : (void)comm;
3240 : (void)ranks;
3241 : (void)useBackup;
3242 : (void)nRanks;
3243 3 : HCCL_ERROR("HcclCommWorkingDevNicSetV2 not support V950.");
3244 1 : return HCCL_E_NOT_SUPPORT;
3245 : }
3246 :
3247 : HcclResult
3248 1 : HcclCommSetMemoryRangeV2(const HcclComm comm, void* baseVirPtr, size_t size, size_t alignment, uint64_t flags)
3249 : {
3250 : (void)comm;
3251 : (void)baseVirPtr;
3252 : (void)size;
3253 : (void)alignment;
3254 : (void)flags;
3255 3 : HCCL_ERROR("HcclCommSetMemoryRangeV2 not support V950.");
3256 1 : return HCCL_E_NOT_SUPPORT;
3257 : }
3258 :
3259 1 : HcclResult HcclCommUnsetMemoryRangeV2(const HcclComm comm, void* baseVirPtr)
3260 : {
3261 : (void)comm;
3262 : (void)baseVirPtr;
3263 3 : HCCL_ERROR("HcclCommUnsetMemoryRangeV2 not support V950.");
3264 1 : return HCCL_E_NOT_SUPPORT;
3265 : }
3266 :
3267 1 : HcclResult HcclCommActivateCommMemoryV2(
3268 : const HcclComm comm, void* virPtr, size_t size, size_t offset, void* handle, uint64_t flags)
3269 : {
3270 : (void)comm;
3271 : (void)virPtr;
3272 : (void)size;
3273 : (void)offset;
3274 : (void)handle;
3275 : (void)flags;
3276 3 : HCCL_ERROR("HcclCommActivateCommMemoryV2 not support V950.");
3277 1 : return HCCL_E_NOT_SUPPORT;
3278 : }
3279 :
3280 1 : HcclResult HcclCommDeactivateCommMemoryV2(const HcclComm comm, void* virPtr)
3281 : {
3282 : (void)comm;
3283 : (void)virPtr;
3284 3 : HCCL_ERROR("HcclCommDeactivateCommMemoryV2 not support V950.");
3285 1 : return HCCL_E_NOT_SUPPORT;
3286 : }
3287 :
3288 0 : uint32_t HcclGetCommConfigCapabilityV2() { return static_cast<uint32_t>(HCCL_COMM_CONFIG_RETRY); }
3289 :
3290 : #ifdef __cplusplus
3291 : }
3292 : #endif // __cplusplus
|
