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 4 : HcclResult HcclCommInitRootInfoConfigV2(
1911 : uint32_t nRanks, const HcclRootInfo* rootInfo, uint32_t rank, const HcclCommConfig* config, HcclComm* comm)
1912 : {
1913 4 : HcclUs startut = TIME_NOW();
1914 4 : CHK_PTR_NULL(rootInfo);
1915 4 : CHK_PTR_NULL(config);
1916 12 : 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 4 : HcclRootHandleV2 rootHandle{};
1921 4 : s32 sRet = memcpy_s(&rootHandle, sizeof(rootHandle), rootInfo->internal, sizeof(rootHandle));
1922 4 : 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 4 : rootHandle.identifier[ROOTINFO_INDENTIFIER_MAX_LENGTH - 1] = '\0';
1929 4 : string identifier = strlen(config->hcclCommName) != 0 ? config->hcclCommName : rootHandle.identifier;
1930 :
1931 : /* 接口交互信息日志 */
1932 12 : 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 4 : CHK_RET(CallSingletons());
1940 4 : HcclCommInfoV2& opbasedCommInfoV2 = GetCommInfoV2();
1941 7 : 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 9 : HCCL_INFO("HcclCommInitRootInfoConfigV2 config->hcclBufferSize[%u] MB", config->hcclBufferSize);
1949 3 : if (UNLIKELY(config->hcclBufferSize == 0)) {
1950 0 : HCCL_ERROR("HcclCommInitRootInfoConfigV2 config: hcclBufferSize is 0 MB, invalid para");
1951 0 : return HCCL_E_PARA;
1952 : }
1953 :
1954 3 : RankTableInfo rankTable{};
1955 3 : RootInfoDetectBridge::DetectContext rootInfoDetectContext;
1956 3 : HcclResult ret = RootInfoDetect(nRanks, rank, rootHandle, rankTable, rootInfoDetectContext);
1957 3 : if (ret != HCCL_SUCCESS) {
1958 3 : HCCL_ERROR(
1959 : "[%s] errNo[0x%016llx] RootInfoDetect failed, identifier[%s].", __func__, HCCL_ERROR_CODE(ret),
1960 : identifier.c_str());
1961 1 : rankTable.Dump();
1962 1 : return ret;
1963 : }
1964 :
1965 : // 打印ranktable
1966 2 : rankTable.Dump();
1967 :
1968 : // 创建通信域
1969 2 : ret = CreateCommConfigRootInfo(rank, config, identifier, rankTable, comm);
1970 2 : CHK_PRT_RET(
1971 : ret,
1972 : HCCL_ERROR(
1973 : "[%s]errNo[0x%016llx] and create comm failed, identifier[%s].", __func__, HCCL_ERROR_CODE(ret),
1974 : identifier.c_str()),
1975 : static_cast<HcclResult>(ret));
1976 :
1977 : /* 关键状态记录 */
1978 6 : HCCL_RUN_INFO(
1979 : "HcclCommInitRootInfoConfigV2 success, take time [%lld]us, rankNum[%u], rank[%u], identifier[%s]",
1980 : DURATION_US(TIME_NOW() - startut), nRanks, rank, identifier.c_str());
1981 2 : return HCCL_SUCCESS;
1982 4 : }
1983 :
1984 2 : HcclResult HcclScatterV2(
1985 : void* sendBuf, void* recvBuf, uint64_t recvCount, HcclDataType dataType, uint32_t root, HcclComm comm,
1986 : aclrtStream stream)
1987 : {
1988 2 : HcclUs startut = TIME_NOW();
1989 : bool isCapture;
1990 2 : rtModel_t rtModel = nullptr;
1991 2 : u32 modelId = 0;
1992 :
1993 2 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
1994 2 : const std::string tag = "Scatter_" + communicator->GetId();
1995 :
1996 2 : CHK_RET_AND_PRINT_IDE(HcomCheckOpParamV2(tag.c_str(), recvCount, dataType, stream), tag.c_str());
1997 2 : u32 rankSize = INVALID_VALUE_RANKSIZE;
1998 2 : CHK_RET_AND_PRINT_IDE(communicator->GetRankSize(&rankSize), tag.c_str());
1999 2 : CHK_RET_AND_PRINT_IDE(HcomCheckUserRankV2(rankSize, root), tag.c_str());
2000 2 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
2001 :
2002 2 : u32 rankId = INVALID_VALUE_RANKID;
2003 2 : CHK_RET(communicator->GetRankId(rankId));
2004 2 : if (rankId == root) { // 本rank为root节点，send_buff不为空
2005 2 : CHK_PTR_NULL(sendBuf);
2006 : }
2007 :
2008 : /* 接口交互信息日志 */
2009 : char stackLogBufferV2[LOG_TMPBUF_SIZE];
2010 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2011 1 : s32 streamId = HrtGetStreamId(stream);
2012 1 : s32 deviceLogicId = HrtGetDevice();
2013 1 : u32 localRank = INVALID_VALUE_RANKID;
2014 1 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
2015 :
2016 1 : s32 ret = snprintf_s(
2017 : stackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
2018 : "tag[%s], sendBuf[%p], recvBuf[%p], recvCount[%llu], dataType[%s], root[%u], localRank[%u], streamId[%d], "
2019 : "deviceLogicId[%d]",
2020 2 : tag.c_str(), sendBuf, recvBuf, recvCount, GetDataTypeEnumStrV2(dataType).c_str(), root, localRank, streamId,
2021 : deviceLogicId);
2022 :
2023 1 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
2024 1 : std::string logInfo = "Entry-HcclScatterV2:" + std::string(stackLogBufferV2);
2025 1 : if (isCapture) {
2026 0 : CHK_PTR_NULL(rtModel);
2027 : // 获取不到modelId会报错
2028 0 : CHK_RET(GetModelId(rtModel, modelId));
2029 0 : logInfo += ", model id[" + to_string(modelId) + "].";
2030 : }
2031 1 : communicator->GetTrace().Save(logInfo);
2032 1 : }
2033 :
2034 2 : static thread_local Hccl::CollOpParams opParams;
2035 2 : opParams.opType = Hccl::OpType::SCATTER;
2036 2 : opParams.dataType = HcclDataTypeToDataType(dataType);
2037 2 : opParams.sendBuf = sendBuf;
2038 2 : opParams.recvBuf = recvBuf;
2039 2 : opParams.count = recvCount;
2040 2 : opParams.root = root;
2041 :
2042 2 : CHK_RET_AND_PRINT_IDE(communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream)), tag.c_str());
2043 :
2044 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2045 1 : HcclUs endut = TIME_NOW();
2046 : /* 关键状态记录 */
2047 2 : std::string endInfo = "HcclScatterV2:success,take time: " + std::to_string(DURATION_US(endut - startut).count())
2048 3 : + " us, tag: " + tag + std::string(stackLogBufferV2);
2049 1 : communicator->GetTrace().Save(endInfo);
2050 1 : }
2051 :
2052 2 : return HCCL_SUCCESS;
2053 2 : }
2054 :
2055 3 : HcclResult HcclAllGatherV2(
2056 : void* sendBuf, void* recvBuf, uint64_t sendCount, HcclDataType dataType, HcclComm comm, aclrtStream stream)
2057 : {
2058 3 : HcclUs startut = TIME_NOW();
2059 : bool isCapture;
2060 3 : rtModel_t rtModel = nullptr;
2061 3 : u32 modelId = 0;
2062 :
2063 3 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
2064 3 : const std::string tag = "AllGather_" + communicator->GetId();
2065 :
2066 3 : CHK_RET_AND_PRINT_IDE(HcomCheckOpParamV2(tag.c_str(), sendCount, dataType, stream), tag.c_str());
2067 3 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
2068 :
2069 : /* 接口交互信息日志 */
2070 : char stackLogBufferV2[LOG_TMPBUF_SIZE];
2071 3 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2072 1 : s32 streamId = HrtGetStreamId(stream);
2073 1 : s32 deviceLogicId = HrtGetDevice();
2074 1 : u32 localRank = INVALID_VALUE_RANKID;
2075 1 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
2076 :
2077 1 : s32 ret = snprintf_s(
2078 : stackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
2079 : "tag[%s], sendBuf[%p], recvBuf[%p], sendCount[%llu], dataType[%s], localRank[%u], streamId[%d],"
2080 : "deviceLogicId[%d]",
2081 2 : tag.c_str(), sendBuf, recvBuf, sendCount, GetDataTypeEnumStrV2(dataType).c_str(), localRank, streamId,
2082 : deviceLogicId);
2083 :
2084 1 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
2085 1 : std::string logInfo = "Entry-HcclAllGatherV2:" + std::string(stackLogBufferV2);
2086 1 : if (isCapture) {
2087 0 : CHK_PTR_NULL(rtModel);
2088 : // 获取不到modelId会报错
2089 0 : CHK_RET(GetModelId(rtModel, modelId));
2090 0 : logInfo += ", model id[" + to_string(modelId) + "].";
2091 : }
2092 1 : communicator->GetTrace().Save(logInfo);
2093 1 : }
2094 :
2095 3 : static thread_local Hccl::CollOpParams opParams;
2096 3 : opParams.opType = Hccl::OpType::ALLGATHER;
2097 3 : opParams.dataType = HcclDataTypeToDataType(dataType);
2098 3 : opParams.sendBuf = sendBuf;
2099 3 : opParams.recvBuf = recvBuf;
2100 3 : opParams.count = sendCount;
2101 3 : opParams.opTag = tag;
2102 :
2103 3 : CHK_RET_AND_PRINT_IDE(communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream)), tag.c_str());
2104 :
2105 3 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2106 1 : HcclUs endut = TIME_NOW();
2107 : /* 关键状态记录 */
2108 : std::string endInfo
2109 2 : = "HcclAllGatherV2:success,take time: " + std::to_string(DURATION_US(endut - startut).count())
2110 3 : + " us, tag: " + tag + std::string(stackLogBufferV2);
2111 1 : communicator->GetTrace().Save(endInfo);
2112 1 : }
2113 :
2114 3 : return HCCL_SUCCESS;
2115 3 : }
2116 :
2117 9 : HcclResult HcclAllGatherVV2(
2118 : void* sendBuf, uint64_t sendCount, void* recvBuf, void* recvCounts, void* recvDispls, HcclDataType dataType,
2119 : HcclComm comm, aclrtStream stream)
2120 : {
2121 9 : HcclUs startut = TIME_NOW();
2122 : bool isCapture;
2123 9 : rtModel_t rtModel = nullptr;
2124 9 : u32 modelId = 0;
2125 :
2126 : // 获取通信域
2127 9 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
2128 9 : const std::string tag = "AllGatherV_" + communicator->GetId();
2129 :
2130 : // 获取rank信息
2131 : uint32_t rankId;
2132 9 : CHK_RET(communicator->GetRankId(rankId));
2133 : uint32_t rankSize;
2134 9 : CHK_RET(communicator->GetRankSize(&rankSize));
2135 : // 参数合法性校验
2136 9 : if (rankSize == 1) {
2137 : // rankSize为1时，退化为AllGather
2138 : // 检查异常回退AGV的情况
2139 3 : if (sendCount == 0) {
2140 3 : HCCL_WARNING("[AllGatherV] sendCount is 0 when single rank");
2141 1 : return HCCL_SUCCESS;
2142 : } else {
2143 5 : CHK_PRT_RET(sendBuf == nullptr, HCCL_ERROR("[AllGatherV] sendBuf is null when single rank"), HCCL_E_PTR);
2144 1 : CHK_PRT_RET(recvBuf == nullptr, HCCL_ERROR("[AllGatherV] recvBuf is null when single rank"), HCCL_E_PTR);
2145 : }
2146 1 : return HcclAllGatherV2(sendBuf, recvBuf, sendCount, dataType, comm, stream);
2147 : }
2148 18 : CHK_RET_AND_PRINT_IDE(HcomCheckOpParamV2(tag.c_str(), sendCount, dataType, stream), tag.c_str());
2149 10 : CHK_RET_AND_PRINT_IDE(HcomCheckVOpParamV2(rankId, rankSize, sendCount, recvCounts), tag.c_str());
2150 3 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
2151 :
2152 : /* 接口交互信息日志 */
2153 : char stackLogBufferV2[LOG_TMPBUF_SIZE];
2154 3 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2155 0 : s32 streamId = HrtGetStreamId(stream);
2156 0 : s32 deviceLogicId = HrtGetDevice();
2157 0 : u32 localRank = INVALID_VALUE_RANKID;
2158 0 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
2159 :
2160 0 : s32 ret = snprintf_s(
2161 : stackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
2162 : "tag[%s], sendBuf[%p], recvBuf[%p], sendCount[%llu], recvCounts[%p], recvDispls[%p], "
2163 : "dataType[%s], localRank[%u], streamId[%d], deviceLogicId[%d]",
2164 0 : tag.c_str(), sendBuf, recvBuf, sendCount, recvCounts, recvDispls, GetDataTypeEnumStrV2(dataType).c_str(),
2165 : localRank, streamId, deviceLogicId);
2166 :
2167 0 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
2168 0 : std::string logInfo = "Entry-HcclAllGatherVV2:" + std::string(stackLogBufferV2);
2169 0 : if (isCapture) {
2170 0 : CHK_PTR_NULL(rtModel);
2171 : // 获取不到modelId会报错
2172 0 : CHK_RET(GetModelId(rtModel, modelId));
2173 0 : logInfo += ", model id[" + to_string(modelId) + "].";
2174 : }
2175 0 : communicator->GetTrace().Save(logInfo);
2176 0 : }
2177 :
2178 3 : u64* counts = static_cast<u64*>(recvCounts);
2179 3 : u64 output = 0;
2180 9 : for (size_t index = 0; index < rankSize; index++) {
2181 6 : output += counts[index];
2182 : }
2183 3 : if (output == 0) {
2184 3 : HCCL_INFO("[%s] output[%llu] is equal to zero", __func__, output);
2185 1 : return HCCL_SUCCESS;
2186 : }
2187 2 : RPT_INPUT_ERR(
2188 : recvBuf == nullptr, "EI0003", std::vector<std::string>({"ccl_op", "value", "parameter", "expect"}),
2189 : std::vector<std::string>({"HcclAllGatherVV2", "nullptr", "recvBuf", "not nullptr"}));
2190 2 : CHK_PTR_NULL(recvBuf);
2191 : // opParams组装
2192 2 : Hccl::CollOpParams opParams;
2193 2 : opParams.opType = Hccl::OpType::ALLGATHERV;
2194 2 : opParams.dataType = HcclDataTypeToDataType(dataType);
2195 2 : opParams.dstRank = rankId;
2196 2 : opParams.sendBuf = sendBuf;
2197 2 : opParams.recvBuf = recvBuf;
2198 2 : opParams.count = sendCount;
2199 2 : opParams.vDataDes.counts = recvCounts;
2200 2 : opParams.vDataDes.displs = recvDispls;
2201 2 : opParams.vDataDes.dataType = HcclDataTypeToDataType(dataType);
2202 :
2203 2 : CHK_RET_AND_PRINT_IDE(communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream)), tag.c_str());
2204 :
2205 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2206 0 : HcclUs endut = TIME_NOW();
2207 : /* 关键状态记录 */
2208 : std::string endInfo
2209 0 : = "HcclAllGatherVV2:success,take time: " + std::to_string(DURATION_US(endut - startut).count())
2210 0 : + " us, tag: " + tag + std::string(stackLogBufferV2);
2211 0 : communicator->GetTrace().Save(endInfo);
2212 0 : }
2213 :
2214 2 : return HCCL_SUCCESS;
2215 9 : }
2216 :
2217 4 : HcclResult ValidateRank(uint32_t rank, Hccl::HcclCommunicator* communicator)
2218 : {
2219 4 : u32 rankSize{};
2220 4 : CHK_RET(communicator->GetRankSize(&rankSize));
2221 4 : u32 rankId{INVALID_VALUE_RANKID};
2222 4 : CHK_RET(communicator->GetRankId(rankId));
2223 4 : CHK_RET(HcomCheckUserRankV2(rankSize, rank));
2224 4 : CHK_PRT_RET(rankId == rank, HCCL_ERROR("same rank[%u] is not allowed", rank), HCCL_E_PARA);
2225 4 : return HCCL_SUCCESS;
2226 : }
2227 :
2228 : HcclResult
2229 2 : HcclSendV2(void* sendBuf, uint64_t count, HcclDataType dataType, uint32_t destRank, HcclComm comm, aclrtStream stream)
2230 : {
2231 2 : HcclUs startut = TIME_NOW();
2232 : bool isCapture;
2233 2 : rtModel_t rtModel = nullptr;
2234 2 : u32 modelId = 0;
2235 :
2236 2 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
2237 2 : const std::string tag = "SendRecv_" + communicator->GetId();
2238 :
2239 2 : CHK_RET(HcomCheckDataTypeV2(dataType));
2240 2 : CHK_RET(ValidateRank(destRank, communicator));
2241 2 : CHK_RET_AND_PRINT_IDE(HcomCheckOpParamV2(tag.c_str(), count, dataType, stream), tag.c_str());
2242 2 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
2243 :
2244 : /* 接口交互信息日志 */
2245 : char hcclSendStackLogBufferV2[LOG_TMPBUF_SIZE];
2246 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2247 1 : s32 streamId = HrtGetStreamId(stream);
2248 1 : s32 deviceLogicId = HrtGetDevice();
2249 1 : u32 localRank = INVALID_VALUE_RANKID;
2250 1 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
2251 :
2252 1 : s32 ret = snprintf_s(
2253 : hcclSendStackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
2254 : "tag[%s], sendBuf[%p], count[%llu], dataType[%s], localRank[%u], streamId[%d], deviceLogicId[%d]",
2255 2 : tag.c_str(), sendBuf, count, GetDataTypeEnumStrV2(dataType).c_str(), localRank, streamId, deviceLogicId);
2256 :
2257 1 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
2258 1 : std::string logInfo = "Entry-HcclSendV2:" + std::string(hcclSendStackLogBufferV2);
2259 1 : if (isCapture) {
2260 0 : CHK_PTR_NULL(rtModel);
2261 : // 获取不到modelId会报错
2262 0 : CHK_RET(GetModelId(rtModel, modelId));
2263 0 : logInfo += ", model id[" + to_string(modelId) + "].";
2264 : }
2265 1 : communicator->GetTrace().Save(logInfo);
2266 1 : }
2267 :
2268 2 : static thread_local Hccl::CollOpParams opParams{};
2269 2 : opParams.opType = Hccl::OpType::SEND;
2270 2 : opParams.dataType = HcclDataTypeToDataType(dataType);
2271 2 : opParams.sendBuf = sendBuf;
2272 2 : opParams.recvBuf = nullptr;
2273 2 : opParams.count = count;
2274 2 : opParams.dstRank = destRank;
2275 2 : opParams.opTag = tag;
2276 :
2277 2 : CHK_RET_AND_PRINT_IDE(communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream)), tag.c_str());
2278 :
2279 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2280 1 : HcclUs endut = TIME_NOW();
2281 : /* 关键状态记录 */
2282 2 : std::string endInfo = "HcclSendV2:success,take time: " + std::to_string(DURATION_US(endut - startut).count())
2283 3 : + " us, tag: " + tag + std::string(hcclSendStackLogBufferV2);
2284 1 : communicator->GetTrace().Save(endInfo);
2285 1 : }
2286 :
2287 2 : return HCCL_SUCCESS;
2288 2 : }
2289 :
2290 : HcclResult
2291 2 : HcclRecvV2(void* recvBuf, uint64_t count, HcclDataType dataType, uint32_t srcRank, HcclComm comm, aclrtStream stream)
2292 : {
2293 2 : HcclUs startut = TIME_NOW();
2294 : bool isCapture;
2295 2 : rtModel_t rtModel = nullptr;
2296 2 : u32 modelId = 0;
2297 :
2298 2 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
2299 2 : const std::string tag = "SendRecv_" + communicator->GetId();
2300 :
2301 2 : CHK_RET(HcomCheckDataTypeV2(dataType));
2302 2 : CHK_RET(ValidateRank(srcRank, communicator));
2303 2 : CHK_RET_AND_PRINT_IDE(HcomCheckOpParamV2(tag.c_str(), count, dataType, stream), tag.c_str());
2304 2 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
2305 :
2306 : /* 接口交互信息日志 */
2307 : char hcclRecvStackLogBufferV2[LOG_TMPBUF_SIZE];
2308 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2309 1 : s32 streamId = HrtGetStreamId(stream);
2310 1 : s32 deviceLogicId = HrtGetDevice();
2311 1 : u32 localRank = INVALID_VALUE_RANKID;
2312 1 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
2313 :
2314 1 : s32 ret = snprintf_s(
2315 : hcclRecvStackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
2316 : "tag[%s], recvBuf[%p], count[%llu], dataType[%s], localRank[%u], streamId[%d], deviceLogicId[%d]",
2317 2 : tag.c_str(), recvBuf, count, GetDataTypeEnumStrV2(dataType).c_str(), localRank, streamId, deviceLogicId);
2318 :
2319 1 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
2320 1 : std::string logInfo = "Entry-HcclRecvV2:" + std::string(hcclRecvStackLogBufferV2);
2321 1 : if (isCapture) {
2322 0 : CHK_PTR_NULL(rtModel);
2323 : // 获取不到modelId会报错
2324 0 : CHK_RET(GetModelId(rtModel, modelId));
2325 0 : logInfo += ", model id[" + to_string(modelId) + "].";
2326 : }
2327 1 : communicator->GetTrace().Save(logInfo);
2328 1 : }
2329 :
2330 2 : static thread_local Hccl::CollOpParams opParams{};
2331 2 : opParams.opType = Hccl::OpType::RECV;
2332 2 : opParams.dataType = HcclDataTypeToDataType(dataType);
2333 2 : opParams.sendBuf = nullptr;
2334 2 : opParams.recvBuf = recvBuf;
2335 2 : opParams.count = count;
2336 2 : opParams.dstRank = srcRank;
2337 2 : opParams.opTag = tag;
2338 :
2339 2 : CHK_RET_AND_PRINT_IDE(communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream)), tag.c_str());
2340 :
2341 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2342 1 : HcclUs endut = TIME_NOW();
2343 : /* 关键状态记录 */
2344 2 : std::string endInfo = "HcclRecvV2:success,take time: " + std::to_string(DURATION_US(endut - startut).count())
2345 3 : + " us, tag: " + tag + std::string(hcclRecvStackLogBufferV2);
2346 1 : communicator->GetTrace().Save(endInfo);
2347 1 : }
2348 :
2349 2 : return HCCL_SUCCESS;
2350 2 : }
2351 :
2352 9 : HcclResult HcclReduceScatterV2(
2353 : void* sendBuf, void* recvBuf, uint64_t recvCount, HcclDataType dataType, HcclReduceOp op, HcclComm comm,
2354 : aclrtStream stream)
2355 : {
2356 9 : HcclUs startut = TIME_NOW();
2357 : bool isCapture;
2358 9 : rtModel_t rtModel = nullptr;
2359 9 : u32 modelId = 0;
2360 :
2361 9 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
2362 9 : const std::string tag = "ReduceScatter_" + communicator->GetId();
2363 :
2364 9 : CHK_RET_AND_PRINT_IDE(HcomCheckOpParamV2(tag.c_str(), recvCount, dataType, stream), tag.c_str());
2365 9 : CHK_RET_AND_PRINT_IDE(HcomCheckReductionOpV2(op), tag.c_str());
2366 21 : CHK_RET_AND_PRINT_IDE(HcomCheckReduceDataTypeV2(dataType, op), tag.c_str());
2367 7 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
2368 :
2369 : /* 接口交互信息日志 */
2370 : char stackLogBufferV2[LOG_TMPBUF_SIZE];
2371 7 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2372 3 : s32 streamId = HrtGetStreamId(stream);
2373 3 : s32 deviceLogicId = HrtGetDevice();
2374 3 : u32 localRank = INVALID_VALUE_RANKID;
2375 3 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
2376 :
2377 6 : s32 ret = snprintf_s(
2378 : stackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
2379 : "tag[%s], sendBuf[%p], recvBuf[%p], recvCount[%llu], dataType[%s], op[%s],"
2380 : "localRank[%u], streamId[%d], deviceLogicId[%d]",
2381 6 : tag.c_str(), sendBuf, recvBuf, recvCount, GetDataTypeEnumStrV2(dataType).c_str(),
2382 6 : GetReduceOpEnumStrV2(op).c_str(), localRank, streamId, deviceLogicId);
2383 :
2384 3 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
2385 3 : std::string logInfo = "Entry-HcclReduceScatterV2:" + std::string(stackLogBufferV2);
2386 3 : if (isCapture) {
2387 0 : CHK_PTR_NULL(rtModel);
2388 : // 获取不到modelId会报错
2389 0 : CHK_RET(GetModelId(rtModel, modelId));
2390 0 : logInfo += ", model id[" + to_string(modelId) + "].";
2391 : }
2392 3 : communicator->GetTrace().Save(logInfo);
2393 3 : }
2394 :
2395 7 : static thread_local Hccl::CollOpParams opParams;
2396 7 : opParams.opType = Hccl::OpType::REDUCESCATTER;
2397 7 : opParams.dataType = HcclDataTypeToDataType(dataType);
2398 7 : opParams.reduceOp = HCCL_OP_REDUCE_MAP[op];
2399 7 : opParams.sendBuf = sendBuf;
2400 7 : opParams.recvBuf = recvBuf;
2401 7 : opParams.count = recvCount;
2402 7 : opParams.opTag = tag;
2403 :
2404 7 : CHK_RET_AND_PRINT_IDE(communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream)), tag.c_str());
2405 :
2406 7 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2407 3 : HcclUs endut = TIME_NOW();
2408 : /* 关键状态记录 */
2409 : std::string endInfo
2410 6 : = "HcclReduceScatterV2:success,take time: " + std::to_string(DURATION_US(endut - startut).count())
2411 9 : + " us, tag: " + tag + std::string(stackLogBufferV2);
2412 3 : communicator->GetTrace().Save(endInfo);
2413 3 : }
2414 :
2415 7 : return HCCL_SUCCESS;
2416 9 : }
2417 :
2418 10 : HcclResult HcclReduceScatterVV2(
2419 : void* sendBuf, void* sendCounts, void* sendDispls, void* recvBuf, uint64_t recvCount, HcclDataType dataType,
2420 : HcclReduceOp op, HcclComm comm, aclrtStream stream)
2421 : {
2422 10 : HcclUs startut = TIME_NOW();
2423 : bool isCapture;
2424 10 : rtModel_t rtModel = nullptr;
2425 10 : u32 modelId = 0;
2426 :
2427 : // 获取通信域
2428 10 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
2429 10 : const std::string tag = "ReduceScatterV_" + communicator->GetId();
2430 :
2431 : // 获取rank信息
2432 : uint32_t rankId;
2433 10 : CHK_RET(communicator->GetRankId(rankId));
2434 : uint32_t rankSize;
2435 10 : CHK_RET(communicator->GetRankSize(&rankSize));
2436 : // 参数合法性校验
2437 10 : if (rankSize == 1) {
2438 : // rankSize为1时，退化为ReduceScatter
2439 : // 检查异常回退RSV的情况
2440 3 : if (recvCount == 0) {
2441 3 : HCCL_WARNING("[ReduceScatterV] recvCount is 0 when single rank");
2442 1 : return HCCL_SUCCESS;
2443 : } else {
2444 5 : CHK_PRT_RET(
2445 : sendBuf == nullptr, HCCL_ERROR("[ReduceScatterV] sendBuf is null when single rank"), HCCL_E_PTR);
2446 1 : CHK_PRT_RET(
2447 : recvBuf == nullptr, HCCL_ERROR("[ReduceScatterV] recvBuf is null when single rank"), HCCL_E_PTR);
2448 : }
2449 1 : return HcclReduceScatterV2(sendBuf, recvBuf, recvCount, dataType, op, comm, stream);
2450 : }
2451 13 : CHK_RET_AND_PRINT_IDE(HcomCheckOpParamV2(tag.c_str(), recvCount, dataType, stream), tag.c_str());
2452 6 : CHK_RET_AND_PRINT_IDE(HcomCheckReductionOpV2(op), tag.c_str());
2453 12 : CHK_RET_AND_PRINT_IDE(HcomCheckReduceDataTypeV2(dataType, op), tag.c_str());
2454 17 : CHK_RET_AND_PRINT_IDE(HcomCheckVOpParamV2(rankId, rankSize, recvCount, sendCounts), tag.c_str());
2455 3 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
2456 :
2457 : /* 接口交互信息日志 */
2458 : char stackLogBufferV2[LOG_TMPBUF_SIZE];
2459 3 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2460 0 : s32 streamId = HrtGetStreamId(stream);
2461 0 : s32 deviceLogicId = HrtGetDevice();
2462 0 : u32 localRank = INVALID_VALUE_RANKID;
2463 0 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
2464 :
2465 0 : s32 ret = snprintf_s(
2466 : stackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
2467 : "tag[%s], sendBuf[%p], recvBuf[%p], sendCounts[%p], sendDispls[%p], recvCount[%llu], dataType[%s], op[%s],"
2468 : "localRank[%u], streamId[%d], deviceLogicId[%d]",
2469 0 : tag.c_str(), sendBuf, recvBuf, sendCounts, sendDispls, recvCount, GetDataTypeEnumStrV2(dataType).c_str(),
2470 0 : GetReduceOpEnumStrV2(op).c_str(), localRank, streamId, deviceLogicId);
2471 :
2472 0 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
2473 0 : std::string logInfo = "Entry-HcclReduceScatterVV2:" + std::string(stackLogBufferV2);
2474 0 : if (isCapture) {
2475 0 : CHK_PTR_NULL(rtModel);
2476 : // 获取不到modelId会报错
2477 0 : CHK_RET(GetModelId(rtModel, modelId));
2478 0 : logInfo += ", model id[" + to_string(modelId) + "].";
2479 : }
2480 0 : communicator->GetTrace().Save(logInfo);
2481 0 : }
2482 :
2483 3 : u64* counts = static_cast<u64*>(sendCounts);
2484 3 : u64 inputCount = 0;
2485 9 : for (size_t index = 0; index < rankSize; index++) {
2486 6 : inputCount += counts[index];
2487 : }
2488 3 : if (inputCount == 0) {
2489 3 : HCCL_INFO("[%s] inputCount[%llu] is equal to zero", __func__, inputCount);
2490 1 : return HCCL_SUCCESS;
2491 : }
2492 2 : RPT_INPUT_ERR(
2493 : sendBuf == nullptr, "EI0003", std::vector<std::string>({"ccl_op", "value", "parameter", "expect"}),
2494 : std::vector<std::string>({"HcclReduceScatterVV2", "nullptr", "sendBuf", "not nullptr"}));
2495 2 : CHK_PTR_NULL(sendBuf);
2496 2 : if (op == HCCL_REDUCE_PROD) {
2497 0 : HCCL_ERROR("[Check][ReductionOp] Op:[HCCL_REDUCE_PROD] not supported, tag[%s]", tag.c_str());
2498 0 : return HCCL_E_NOT_SUPPORT;
2499 : }
2500 : // opParams组装
2501 2 : Hccl::CollOpParams opParams;
2502 2 : opParams.opType = Hccl::OpType::REDUCESCATTERV;
2503 2 : opParams.dataType = HcclDataTypeToDataType(dataType);
2504 2 : opParams.reduceOp = HcclReduceOpToReduceOp(op);
2505 2 : opParams.dstRank = rankId;
2506 2 : opParams.sendBuf = sendBuf;
2507 2 : opParams.recvBuf = recvBuf;
2508 2 : opParams.count = recvCount;
2509 2 : opParams.vDataDes.counts = sendCounts;
2510 2 : opParams.vDataDes.displs = sendDispls;
2511 2 : opParams.vDataDes.dataType = HcclDataTypeToDataType(dataType);
2512 :
2513 2 : CHK_RET_AND_PRINT_IDE(communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream)), tag.c_str());
2514 :
2515 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2516 0 : HcclUs endut = TIME_NOW();
2517 : /* 关键状态记录 */
2518 : std::string endInfo
2519 0 : = "HcclReduceScatterVV2:success,take time: " + std::to_string(DURATION_US(endut - startut).count())
2520 0 : + " us, tag: " + tag + std::string(stackLogBufferV2);
2521 0 : communicator->GetTrace().Save(endInfo);
2522 0 : }
2523 :
2524 2 : return HCCL_SUCCESS;
2525 10 : }
2526 :
2527 2 : HcclResult HcclBatchSendRecvV2(HcclSendRecvItem* sendRecvInfo, uint32_t itemNum, HcclComm comm, aclrtStream stream)
2528 : {
2529 2 : HcclUs startut = TIME_NOW();
2530 : bool isCapture;
2531 2 : rtModel_t rtModel = nullptr;
2532 2 : u32 modelId = 0;
2533 :
2534 2 : CHK_PTR_NULL(comm);
2535 2 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
2536 2 : const std::string tag = "HcclBatchSendRecvV2_" + communicator->GetId();
2537 2 : CHK_PTR_NULL(stream);
2538 2 : CHK_PTR_NULL(sendRecvInfo);
2539 2 : CHK_PRT_RET(itemNum == 0, HCCL_WARNING("[BatchSendRecv] taskList itemNum is zero."), HCCL_SUCCESS);
2540 2 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
2541 2 : u32 rankSize = 0;
2542 2 : CHK_RET(communicator->GetRankSize(&rankSize)); // 获取rankSize, 后续需要校验userrank是否在0到rankSize-1之间
2543 22 : for (uint32_t i = 0; i < itemNum; i++) {
2544 20 : if ((sendRecvInfo + i)->buf == nullptr) {
2545 11 : continue; // 支持数据量为0的场景，buf为空的跳过
2546 : }
2547 9 : CHK_RET_AND_PRINT_IDE(
2548 : HcomCheckOpParamV2(tag.c_str(), (sendRecvInfo + i)->count, (sendRecvInfo + i)->dataType, stream),
2549 : tag.c_str());
2550 9 : CHK_RET(HcomCheckUserRankV2(rankSize, (sendRecvInfo + i)->remoteRank));
2551 : }
2552 :
2553 : /* 记录接口交互信息日志 */
2554 : char stackLogBufferV2[LOG_TMPBUF_SIZE];
2555 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2556 1 : s32 streamId = HrtGetStreamId(stream);
2557 1 : s32 deviceLogicId = HrtGetDevice();
2558 1 : u32 localRank = INVALID_VALUE_RANKID;
2559 1 : CHK_RET_AND_PRINT_IDE(communicator->GetRankId(localRank), tag.c_str());
2560 :
2561 1 : s32 ret = snprintf_s(
2562 : stackLogBufferV2, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
2563 : "tag[%s], itemNum[%u], localRank[%u], streamId[%d], deviceLogicId[%d]", tag.c_str(), itemNum, localRank,
2564 : streamId, deviceLogicId);
2565 :
2566 1 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", tag.c_str()));
2567 1 : std::string logInfo = "Entry-HcclBatchSendRecvV2:" + std::string(stackLogBufferV2);
2568 1 : if (isCapture) {
2569 0 : CHK_PTR_NULL(rtModel);
2570 0 : CHK_RET(GetModelId(rtModel, modelId));
2571 0 : logInfo += ", model id[" + to_string(modelId) + "].";
2572 : }
2573 1 : communicator->GetTrace().Save(logInfo);
2574 1 : }
2575 :
2576 2 : static thread_local Hccl::CollOpParams opParams;
2577 2 : opParams.opType = Hccl::OpType::BATCHSENDRECV;
2578 2 : opParams.batchSendRecvDataDes.sendRecvItemsPtr = static_cast<void*>(sendRecvInfo);
2579 2 : opParams.batchSendRecvDataDes.itemNum = itemNum;
2580 2 : opParams.dataType = HcclDataTypeToDataType(sendRecvInfo->dataType);
2581 :
2582 2 : CHK_RET_AND_PRINT_IDE(communicator->LoadOpbasedCollOp(opParams, static_cast<void*>(stream)), tag.c_str());
2583 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2584 1 : HcclUs endut = TIME_NOW();
2585 : std::string endInfo
2586 2 : = "HcclBatchSendRecvV2:success,take time: " + std::to_string(DURATION_US(endut - startut).count())
2587 3 : + " us, tag: " + tag + std::string(stackLogBufferV2);
2588 1 : communicator->GetTrace().Save(endInfo);
2589 1 : }
2590 2 : return HCCL_SUCCESS;
2591 2 : }
2592 :
2593 : // 功能说明：推动式建链，超时退出
2594 2 : HcclResult WaitAllCommReady(s32 deviceLogicId)
2595 : {
2596 : try {
2597 2 : HrtSetDevice(deviceLogicId);
2598 2 : HcclCommInfoV2& opbasedCommInfoV2 = GetCommInfoV2();
2599 : // 定义最大等待10秒
2600 2 : constexpr u32 waitTransportReadyTimeoutMs = 10 * 1000; // 待解决
2601 2 : auto timeout = std::chrono::milliseconds(waitTransportReadyTimeoutMs);
2602 2 : HcclUs startTime = std::chrono::steady_clock::now();
2603 :
2604 : // 创建锁，防止在建链过程中，依旧还在恢复通信域
2605 2 : std::unique_lock<std::mutex> groupParaLock(opbasedCommInfoV2.groupParamsLock);
2606 6 : HCCL_INFO("[%s] deviceLogicId[%d] start wait all comm ready", __func__, deviceLogicId);
2607 : // 轮巡调度，推动式建链
2608 : while (true) {
2609 2 : bool isAllCommReady = true;
2610 : Hccl::HcclCommunicator* communicator;
2611 : // 枚举所有通信域进行推动式建链
2612 5 : for (auto iter = opbasedCommInfoV2.hcclGroupMap.begin(); iter != opbasedCommInfoV2.hcclGroupMap.end();
2613 3 : iter++) {
2614 3 : CHK_PTR_NULL(iter->second.pComm);
2615 3 : communicator = static_cast<Hccl::HcclCommunicator*>(iter->second.pComm.get());
2616 3 : if (!communicator->IsCommReady()) {
2617 : // 只要任意Comm一个没有ready，整体建链结果为 false
2618 1 : isAllCommReady = false;
2619 : }
2620 : }
2621 2 : if (isAllCommReady) {
2622 1 : break;
2623 : }
2624 :
2625 : // 超时判断
2626 1 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
2627 3 : HCCL_ERROR("[%s]WaitAllCommReady timeout, deviceLogicId[%d].", __func__, deviceLogicId);
2628 1 : return HCCL_E_TIMEOUT;
2629 : }
2630 0 : }
2631 1 : HrtResetDevice(deviceLogicId);
2632 2 : } catch (HcclException& e) {
2633 0 : HCCL_ERROR(e.what());
2634 0 : return e.GetErrorCode();
2635 0 : } catch (std::exception& e) {
2636 0 : HCCL_ERROR(e.what());
2637 0 : return HCCL_E_INTERNAL;
2638 0 : } catch (...) {
2639 0 : HCCL_ERROR("Unknown error occurs!");
2640 0 : return HCCL_E_INTERNAL;
2641 0 : }
2642 :
2643 3 : HCCL_INFO("[%s] all comm ready.", __func__);
2644 1 : return HCCL_SUCCESS;
2645 : }
2646 :
2647 : #ifdef __cplusplus
2648 : extern "C" {
2649 : #endif // __cplusplus
2650 :
2651 : HcclResult
2652 0 : HcclGetCclBuffer(HcclComm comm, uintptr_t& cclBufferAddr, size_t& cclBufferSize, HcclMemType& cclBufferMemType)
2653 : {
2654 0 : Hccl::HcclCommunicator* communicatorV2 = (static_cast<Hccl::HcclCommunicator*>(comm));
2655 0 : CHK_PTR_NULL(communicatorV2);
2656 0 : CHK_RET(communicatorV2->HcclGetCclBuffer(cclBufferAddr, cclBufferSize, cclBufferMemType));
2657 0 : return HCCL_SUCCESS;
2658 : }
2659 :
2660 1 : HcclResult HcclGetRawCommHandle(const char* commName, HcclComm* commHandle)
2661 : {
2662 1 : CHK_PTR_NULL(commName);
2663 1 : CHK_PTR_NULL(commHandle);
2664 :
2665 1 : HcclCommInfoV2& opbasedCommInfoV2 = GetCommInfoV2();
2666 1 : std::unique_lock<std::mutex> lock(opbasedCommInfoV2.groupParamsLock);
2667 3 : HCCL_INFO("[HcclGetRawCommHandle] group:[%s]", commName);
2668 2 : auto iter = opbasedCommInfoV2.hcclGroupMap.find(commName);
2669 1 : if (iter == opbasedCommInfoV2.hcclGroupMap.end()) {
2670 0 : HCCL_ERROR("[HcclGetRawCommHandle] commName [%s] not found, please check.", commName);
2671 0 : return HCCL_E_PARA;
2672 : }
2673 2 : *commHandle = static_cast<HcclComm>(opbasedCommInfoV2.hcclGroupMap[commName].pComm.get());
2674 1 : return HCCL_SUCCESS;
2675 1 : }
2676 :
2677 2 : HcclResult HcclGetCcuTaskInfoLegacy(HcclComm comm, void* tilingData, void* ccuTaskGroup)
2678 : {
2679 2 : CHK_PTR_NULL(comm);
2680 2 : CHK_PTR_NULL(tilingData);
2681 2 : CHK_PTR_NULL(ccuTaskGroup);
2682 :
2683 2 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
2684 2 : if (EnvConfig::GetInstance().GetLogConfig().GetEntryLogEnable()) {
2685 0 : HCCL_RUN_INFO(
2686 : "Entry-HcclGetCcuTaskInfo V950, commId[%s], tilingData[%p], ccuTaskGroup[%p]",
2687 : communicator->GetId().c_str(), tilingData, ccuTaskGroup);
2688 : }
2689 2 : auto ret = communicator->GetCcuTaskInfo(tilingData, ccuTaskGroup);
2690 2 : if (ret != HCCL_SUCCESS) {
2691 3 : HCCL_ERROR("HcclGetCcuTaskInfo ret[%d] commId[%s]", ret, communicator->GetId().c_str());
2692 1 : return HCCL_E_INTERNAL;
2693 : }
2694 :
2695 1 : return HCCL_SUCCESS;
2696 : }
2697 :
2698 2 : HcclResult HcclSnapshotSave(void* snapshotBuf, uint32_t size, uint32_t step)
2699 : {
2700 : // 快照保存只支持ranktable场景，不支持topo探测场景
2701 6 : HCCL_INFO("[%s] snapshot save start, size[%u], step[%u]", __func__, size, step);
2702 2 : CHK_PTR_NULL(snapshotBuf);
2703 : // 获取公共信息
2704 2 : HcclCommInfoV2& opbasedCommInfoV2 = GetCommInfoV2();
2705 : // 获取公共信息里的step信息
2706 2 : u64 savedStep = opbasedCommInfoV2.step;
2707 : // 如果当前step不等于用户传入的step，直接返回报错
2708 2 : if (savedStep != step) {
2709 0 : HCCL_ERROR("[%s] step is not match, savedStep[%u], userInputStep[%u]", __func__, savedStep, step);
2710 0 : return HCCL_E_PARA;
2711 : }
2712 : // 获取保存的快照
2713 2 : Hccl::BinaryStream& savedSnapshotBuf = Hccl::SnapShotParser::GetInstance().GetSnapShotBuf();
2714 2 : uint32_t dataLen = savedSnapshotBuf.GetSize();
2715 6 : HCCL_INFO("[%s] savedSnapshotBuf data len[%u]", __func__, dataLen);
2716 :
2717 : // 获取保存的size
2718 2 : if ((dataLen + sizeof(dataLen) + sizeof(uint32_t)) != size) {
2719 : // 如果当前size不等于用户传入的size，直接返回报错
2720 3 : HCCL_ERROR("[%s] size is not match, userInputSize[%u] dataLen[%u]", __func__, size, dataLen);
2721 1 : return HCCL_E_PARA;
2722 : }
2723 1 : std::vector<char> data;
2724 1 : savedSnapshotBuf.DumpWithRevert(data);
2725 3 : HCCL_INFO("[%s] dump data size [%u]", __func__, data.size());
2726 1 : if (data.empty()) {
2727 3 : HCCL_INFO("[%s] dump data empty", __func__);
2728 1 : return HCCL_E_INTERNAL;
2729 : }
2730 :
2731 : // 计算快照crc值
2732 0 : uint32_t crcValue{0};
2733 0 : CHK_RET(Hccl::SnapShotParser::GetInstance().CalcBufCrc32(savedSnapshotBuf, crcValue));
2734 :
2735 : // 将快照大小拷贝到用户传入的内存中
2736 0 : s32 sRet = memcpy_s(snapshotBuf, size, static_cast<void*>(&dataLen), sizeof(dataLen));
2737 0 : CHK_PRT_RET(sRet != EOK, HCCL_ERROR("[%s] memcpy dataLen failed, return[%d]", __func__, sRet), HCCL_E_MEMORY);
2738 : // 将快照crc值拷贝到用户传入的内存中
2739 0 : sRet = memcpy_s(
2740 0 : static_cast<char*>(snapshotBuf) + sizeof(dataLen), size - sizeof(dataLen), static_cast<void*>(&crcValue),
2741 : sizeof(crcValue));
2742 0 : CHK_PRT_RET(sRet != EOK, HCCL_ERROR("[%s] memcpy crcValue failed, return[%d]", __func__, sRet), HCCL_E_MEMORY);
2743 : // 将快照拷贝到用户传入的内存中
2744 0 : sRet = memcpy_s(static_cast<char*>(snapshotBuf) + sizeof(dataLen) + sizeof(crcValue), dataLen, &data[0], dataLen);
2745 0 : CHK_PRT_RET(sRet != EOK, HCCL_ERROR("[%s] memcpy data failed, return[%d]", __func__, sRet), HCCL_E_MEMORY);
2746 0 : return HCCL_SUCCESS;
2747 1 : }
2748 :
2749 1 : void RecoverSnapshotCcuStatus(const std::shared_ptr<Hccl::SnapShotBuf>& savedSnapshotBuf)
2750 : {
2751 1 : HcclCommInfoV2& opbasedCommInfoV2 = GetCommInfoV2();
2752 : // 通过进程锁看护，避免多个通信域同时占用CCU_MS
2753 1 : std::unique_lock<std::mutex> lock(opbasedCommInfoV2.groupParamsLock);
2754 :
2755 1 : for (auto useMsCommId : savedSnapshotBuf->ccuStatusSnapshot.useMsCommIds) {
2756 0 : opbasedCommInfoV2.ccuStatus.useMsCommIds.push_back(useMsCommId.data());
2757 : }
2758 :
2759 1 : for (auto useSchedCommId : savedSnapshotBuf->ccuStatusSnapshot.useSchedCommIds) {
2760 0 : opbasedCommInfoV2.ccuStatus.useSchedCommIds.push_back(useSchedCommId.data());
2761 : }
2762 : // 结果打印
2763 3 : HCCL_INFO(
2764 : "RecoverSnapshotCcuStatus ccuMsCommIds size[%lu] ccuSchedCommIds size[%lu]",
2765 : opbasedCommInfoV2.ccuStatus.useMsCommIds.size(), opbasedCommInfoV2.ccuStatus.useSchedCommIds.size());
2766 1 : for (auto useMsCommId : opbasedCommInfoV2.ccuStatus.useMsCommIds) {
2767 0 : HCCL_DEBUG("RecoverSnapshotCcuStatus useMsCommId[%s]", useMsCommId.c_str());
2768 0 : }
2769 1 : for (auto useSchedCommId : opbasedCommInfoV2.ccuStatus.useSchedCommIds) {
2770 0 : HCCL_DEBUG("RecoverSnapshotCcuStatus useSchedCommId[%s]", useSchedCommId.c_str());
2771 0 : }
2772 1 : }
2773 :
2774 1 : HcclResult HcclSnapshotRecoverAllComms(
2775 : const char* clusterInfo, const char* changedInfo, void* snapshotBuf, uint32_t snapshotBufSize)
2776 : {
2777 : (void)clusterInfo;
2778 : // 入参校验，clusterInfo和changedInfo暂时用不到，先不判空
2779 3 : HCCL_INFO("[%s] snapshot recover start.", __func__);
2780 1 : CHK_PTR_NULL(snapshotBuf);
2781 :
2782 : // 当前不支持在不下发算子的情况下重复调用recover接口
2783 1 : if (Hccl::SnapShotParser::GetInstance().GetIsNeedLoadOp()) {
2784 0 : HCCL_ERROR("[%s] snapshot recover failed, it is necessary to use it after load op, please check!", __func__);
2785 0 : return HCCL_E_INTERNAL;
2786 : }
2787 :
2788 1 : s32 deviceLogicId = HcclGetThreadDeviceId();
2789 :
2790 1 : CHK_RET(CallSingletons()); // 临时规避，在初始化通信域前声明单例保证时序
2791 :
2792 : // 获取公共信息
2793 1 : HcclCommInfoV2& opbasedCommInfoV2 = GetCommInfoV2();
2794 : // 调用恢复快照公共信息函数，让字节流的struct数据恢复到savedSnapshotBuf
2795 1 : std::shared_ptr<Hccl::SnapShotBuf> savedSnapshotBuf = std::make_shared<Hccl::SnapShotBuf>();
2796 1 : CHK_RET(
2797 : Hccl::SnapShotParser::GetInstance().ParseSnapshotToLocalBuff(snapshotBuf, snapshotBufSize, *savedSnapshotBuf));
2798 :
2799 1 : opbasedCommInfoV2.step = savedSnapshotBuf->snapShotPub.step;
2800 : // 将状态设置为RECOVERED
2801 1 : opbasedCommInfoV2.status = DeviceStatus::DEVICE_RECOVERED;
2802 :
2803 : // 创建通信域
2804 1 : opbasedCommInfoV2.pComm.reset(new (std::nothrow) Hccl::HcclCommunicator(
2805 1 : savedSnapshotBuf->snapshot.snapShotComm.commParams, &savedSnapshotBuf->snapshot.snapShotComm.config));
2806 1 : CHK_PTR_NULL(opbasedCommInfoV2.pComm);
2807 1 : opbasedCommInfoV2.pComm->RegisterAcceStateCallBack(CommunicatorCallback());
2808 : // 恢复全局通讯域
2809 3 : HCCL_INFO("[%s] recover global group.", __func__);
2810 1 : CHK_RET(opbasedCommInfoV2.pComm->RecoverComm(
2811 : static_cast<void*>(&savedSnapshotBuf->snapshot.snapShotComm), savedSnapshotBuf->snapShotPub.step, changedInfo));
2812 3 : HCCL_INFO("[%s] global group recover success.", __func__);
2813 1 : opbasedCommInfoV2.commParams = savedSnapshotBuf->snapshot.snapShotComm.commParams;
2814 :
2815 1 : HcclGroupParamsV2 params{};
2816 1 : params.pComm = opbasedCommInfoV2.pComm;
2817 2 : opbasedCommInfoV2.hcclGroupMap[savedSnapshotBuf->snapshot.groupName] = params;
2818 :
2819 : // 这里变成多线程后，封装成小函数
2820 2 : for (uint32_t index = 0; index < savedSnapshotBuf->groupNum; index++) {
2821 1 : std::string groupName(savedSnapshotBuf->subSnapshot[index].groupName);
2822 : // 别重复恢复全局通讯域
2823 1 : if (groupName == savedSnapshotBuf->snapshot.groupName) {
2824 0 : continue;
2825 : }
2826 1 : const Hccl::SnapShotSubComm* snapShotSubComm = &savedSnapshotBuf->subSnapshot[index].snapShotSubComm;
2827 : // 创建一个指针
2828 : std::shared_ptr<Hccl::HcclCommunicator> commImp
2829 1 : = make_shared<Hccl::HcclCommunicator>(snapShotSubComm->commParams, &snapShotSubComm->config);
2830 : // 循环创建恢复子通讯域
2831 3 : HCCL_INFO("[%s] recover sub group[%s].", __func__, groupName.c_str());
2832 1 : CHK_RET(opbasedCommInfoV2.pComm->RecoverSubComm(
2833 : static_cast<const void*>(snapShotSubComm), commImp, savedSnapshotBuf->snapShotPub.step));
2834 3 : HCCL_INFO("[%s] sub group[%s] recover success.", __func__, groupName.c_str());
2835 1 : HcclGroupParamsV2 params{};
2836 1 : params.pComm = commImp;
2837 : // 这里进行多线程后，加锁
2838 1 : opbasedCommInfoV2.hcclGroupMap[groupName] = params;
2839 1 : }
2840 1 : RecoverSnapshotCcuStatus(savedSnapshotBuf);
2841 3 : HCCL_INFO("[%s] snapshot recover success.", __func__);
2842 1 : std::unique_ptr<std::thread> waitReadyThread;
2843 1 : waitReadyThread.reset(new (std::nothrow) std::thread(&WaitAllCommReady, deviceLogicId));
2844 1 : if (waitReadyThread == nullptr) {
2845 0 : HCCL_ERROR("[%s] new waitReadyThread failed.", __func__);
2846 0 : return HCCL_E_INTERNAL;
2847 : }
2848 : // 异步方案目前未分析清楚，采用临时方案规避
2849 1 : waitReadyThread->join();
2850 1 : Hccl::SnapShotParser::GetInstance().SetIsNeedLoadOp(true);
2851 3 : HCCL_INFO("[%s] all comm ready.", __func__);
2852 1 : return HCCL_SUCCESS;
2853 1 : }
2854 3 : static HcclResult GetAllSnapShotStaticBuf(
2855 : const std::shared_ptr<Hccl::HcclCommunicator>& pComm,
2856 : const std::map<std::string, std::shared_ptr<Hccl::HcclCommunicator>>& hcclGroupMap, uint32_t step,
2857 : Hccl::BinaryStream& buf)
2858 : {
2859 9 : HCCL_INFO("[%s] start", __func__);
2860 3 : if (!pComm->IsWorldGroup()) {
2861 0 : HCCL_ERROR("[%s] input comm is not hccl_world_group, please check!", __func__);
2862 0 : return HCCL_E_INTERNAL;
2863 : }
2864 : // 生成 静态流公共数据,直接往buf写
2865 3 : Hccl::SnapShotParser::GetInstance().SerializeCommVersionInfo(buf);
2866 : // 存储 全局通信域名称
2867 3 : buf << pComm->GetId();
2868 : // 存储 全局通信域 静态buf
2869 3 : Hccl::BinaryStream& pCommBuf = *(static_cast<Hccl::BinaryStream*>(pComm->GetStaticBinaryInfo()));
2870 3 : std::vector<char> pCommChars{};
2871 3 : pCommBuf.Dump(pCommChars);
2872 3 : for (auto c : pCommChars) {
2873 0 : buf << c;
2874 : }
2875 :
2876 3 : size_t pSubCommSize = hcclGroupMap.size() - 1;
2877 9 : HCCL_INFO("[%s] pSubCommSize[%u]", __func__, pSubCommSize);
2878 3 : buf << pSubCommSize; // 子通信域 静态buf 数量
2879 :
2880 3 : auto iter = hcclGroupMap.begin();
2881 6 : for (; iter != hcclGroupMap.end(); iter++) {
2882 3 : auto pSubCommName = iter->first;
2883 3 : if (pComm->GetId() == pSubCommName) {
2884 2 : continue;
2885 : }
2886 1 : buf << pSubCommName; // 子通信域 名字
2887 : // 获取子通信域 静态buf
2888 1 : auto groupComm = iter->second;
2889 1 : Hccl::BinaryStream& pSubCommBuf = *(static_cast<Hccl::BinaryStream*>(groupComm->GetStaticBinaryInfo()));
2890 1 : std::vector<char> pSubCommChars{};
2891 1 : pSubCommBuf.Dump(pSubCommChars);
2892 1 : for (auto c : pSubCommChars) {
2893 0 : buf << c;
2894 : }
2895 3 : }
2896 3 : buf << step;
2897 9 : HCCL_INFO("[%s] end", __func__);
2898 :
2899 3 : return HCCL_SUCCESS;
2900 3 : }
2901 :
2902 3 : static HcclResult GetAllSnapShotDynamicBuf(
2903 : const std::shared_ptr<Hccl::HcclCommunicator>& pComm,
2904 : const std::map<std::string, std::shared_ptr<Hccl::HcclCommunicator>>& hcclGroupMap, Hccl::BinaryStream& buf)
2905 : {
2906 9 : HCCL_INFO("[%s] start", __func__);
2907 3 : CHK_PTR_NULL(pComm);
2908 : // 全局通信域 动态buf信息
2909 3 : CHK_RET(pComm->GetSnapShotDynamicBuf(static_cast<void*>(&buf)));
2910 :
2911 3 : size_t pSubCommSize = hcclGroupMap.size() - 1;
2912 3 : buf << pSubCommSize; // 子通信域 动态buf信息 数量
2913 9 : HCCL_INFO("[%s] hcclGroupMap size[%u]", __func__, hcclGroupMap.size());
2914 3 : auto iter = hcclGroupMap.begin();
2915 6 : for (; iter != hcclGroupMap.end(); iter++) {
2916 3 : auto pSubCommName = iter->first;
2917 3 : if (pComm->GetId() == pSubCommName) {
2918 2 : continue;
2919 : }
2920 : // 获取子通信域 建链邻居信息buf
2921 1 : CHK_RET(iter->second->GetSnapShotDynamicBuf(static_cast<void*>(&buf)));
2922 3 : }
2923 9 : HCCL_INFO("[%s] end", __func__);
2924 3 : return HCCL_SUCCESS;
2925 : }
2926 :
2927 3 : void GetSnapShotCcuStatusBuf(Hccl::BinaryStream& buf)
2928 : {
2929 3 : HcclCommInfoV2& opbasedCommInfoV2 = GetCommInfoV2();
2930 : // 通过进程锁看护，避免多个通信域同时占用CCU_MS
2931 3 : std::unique_lock<std::mutex> lock(opbasedCommInfoV2.groupParamsLock);
2932 :
2933 3 : auto ccuStatus = opbasedCommInfoV2.ccuStatus;
2934 3 : buf << ccuStatus.useMsCommIds.size();
2935 9 : HCCL_INFO("useMsCommIds size is %u", ccuStatus.useMsCommIds.size());
2936 3 : for (auto useMsCommId : ccuStatus.useMsCommIds) {
2937 0 : buf << useMsCommId;
2938 0 : HCCL_INFO("useMsCommId is %s", useMsCommId.c_str());
2939 0 : }
2940 :
2941 3 : buf << ccuStatus.useSchedCommIds.size();
2942 9 : HCCL_INFO("useSchedCommIds size is %u", ccuStatus.useSchedCommIds.size());
2943 3 : for (auto useSchedCommId : ccuStatus.useSchedCommIds) {
2944 0 : buf << useSchedCommId;
2945 0 : HCCL_INFO("useSchedCommId is %s", useSchedCommId.c_str());
2946 0 : }
2947 3 : }
2948 :
2949 : // 获取快照占用buffer的大小 --给op_base API getSize调，内部生成完整长流。save后释放
2950 3 : HcclResult SnapshotGenerate(
2951 : const std::shared_ptr<Hccl::HcclCommunicator>& pComm,
2952 : const std::map<std::string, std::shared_ptr<Hccl::HcclCommunicator>>& hcclGroupMap, uint32_t step, uint32_t* size)
2953 : {
2954 3 : CHK_PTR_NULL(size);
2955 3 : CHK_PTR_NULL(pComm);
2956 9 : HCCL_INFO("[%s] start", __func__);
2957 3 : Hccl::SnapShotParser::GetInstance().GetSnapShotBuf().Clear();
2958 3 : CHK_RET(GetAllSnapShotStaticBuf(pComm, hcclGroupMap, step, Hccl::SnapShotParser::GetInstance().GetSnapShotBuf()));
2959 3 : CHK_RET(GetAllSnapShotDynamicBuf(pComm, hcclGroupMap, Hccl::SnapShotParser::GetInstance().GetSnapShotBuf()));
2960 3 : GetSnapShotCcuStatusBuf(Hccl::SnapShotParser::GetInstance().GetSnapShotBuf());
2961 : // size = 流长度 + crc（u32）长度 + 存储流长度所需长度
2962 3 : uint32_t dataLen = static_cast<uint32_t>(Hccl::SnapShotParser::GetInstance().GetSnapShotBuf().GetSize());
2963 3 : *size = dataLen + sizeof(dataLen) + sizeof(uint32_t); // 快照头上保存一个总长度和crc长度
2964 9 : HCCL_INFO("[%s] end, size[%u]", __func__, *size);
2965 3 : return HCCL_SUCCESS;
2966 : }
2967 :
2968 2 : HcclResult HcclSnapshotGetBufSize(uint32_t step, uint32_t* size)
2969 : {
2970 : // 校验DevType
2971 6 : HCCL_INFO("[%s] start", __func__);
2972 2 : Hccl::DevType devType = HrtGetDeviceType();
2973 2 : if (devType != DevType::DEV_TYPE_950 && devType != DevType::DEV_TYPE_960) {
2974 0 : HCCL_INFO("[%s] Get buffer size not support in this device type[%d]", __func__, devType);
2975 0 : return HCCL_E_NOT_SUPPORT;
2976 : }
2977 :
2978 : // step保存在g_opbasedCommInfoV2，save的时候校验，防止GetBufSize和Save的step不一致
2979 2 : HcclCommInfoV2& opbasedCommInfoV2 = GetCommInfoV2();
2980 2 : opbasedCommInfoV2.step = step;
2981 :
2982 2 : std::map<std::string, std::shared_ptr<Hccl::HcclCommunicator>> subCommMap;
2983 4 : for (auto hcclGroupMap : opbasedCommInfoV2.hcclGroupMap) {
2984 2 : subCommMap.insert(std::make_pair(hcclGroupMap.first, hcclGroupMap.second.pComm));
2985 2 : }
2986 2 : return SnapshotGenerate(opbasedCommInfoV2.pComm, subCommMap, step, size);
2987 2 : }
2988 :
2989 1 : HcclResult HcclGetTopoDescV2()
2990 : {
2991 3 : HCCL_ERROR("Current chip type does not support GetTopoDesc.");
2992 1 : return HCCL_E_NOT_SUPPORT;
2993 : }
2994 :
2995 1 : HcclResult HcclGetCommAsyncErrorV2() { return HCCL_SUCCESS; }
2996 :
2997 0 : HcclResult HcclSetConfigV2(HcclConfig config, HcclConfigValue configValue)
2998 : {
2999 : (void)(config);
3000 : (void)(configValue);
3001 0 : HCCL_WARNING("DETERMINISTIC_ENABLE is default option in 950! Can not set.");
3002 0 : return HCCL_SUCCESS;
3003 : }
3004 0 : HcclResult HcclGetConfigV2(HcclConfig config, HcclConfigValue* configValue)
3005 : {
3006 : (void)(config);
3007 0 : constexpr int32_t DETERMINISTIC_ENABLE = 1; // A5支持确定性，不需要配置
3008 0 : (*configValue).value = DETERMINISTIC_ENABLE;
3009 0 : HCCL_WARNING("DETERMINISTIC_ENABLE is default option in 950!");
3010 0 : return HCCL_SUCCESS;
3011 : }
3012 :
3013 1 : HcclResult HcclGetRankGraphV2(const HcclComm* comm, void** rankGraph)
3014 : {
3015 1 : Hccl::HcclCommunicator* communicatorV2 = (static_cast<Hccl::HcclCommunicator*>(*comm));
3016 1 : communicatorV2->GetRankGraphV2(*rankGraph);
3017 1 : return HCCL_SUCCESS;
3018 : }
3019 :
3020 0 : HcclResult HcommFlushV2()
3021 : {
3022 0 : HCCL_INFO("[HcommFlushV2]");
3023 0 : return FlushManager::GetInstance().Flush();
3024 : }
3025 :
3026 0 : HcclResult CommGetCCLBufSizeCfgV2(HcclComm comm, uint64_t* cclBufSize)
3027 : {
3028 0 : HCCL_RUN_INFO("Entry-CommGetCCLBufSizeCfg V950");
3029 0 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
3030 0 : CHK_RET(communicator->GetConfigInCCLbufferSize(cclBufSize));
3031 0 : return HCCL_SUCCESS;
3032 : }
3033 :
3034 0 : HcclResult HcclGetNetLayersV2(HcclComm comm, uint32_t** netLayers, uint32_t* netLayerNum)
3035 : {
3036 0 : HCCL_RUN_INFO("Entry-HcclGetNetLayersV2 V950");
3037 0 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
3038 0 : auto ret = communicator->GetNetLayers(netLayers, netLayerNum);
3039 0 : if (ret != HCCL_SUCCESS) {
3040 0 : HCCL_ERROR(
3041 : "HcclGetNetLayersV2 get netLayers from communicator failed, commId[%s], ret[%d]",
3042 : communicator->GetId().c_str(), ret);
3043 0 : return HCCL_E_NOT_FOUND; // 查询失败返回
3044 : }
3045 0 : return HCCL_SUCCESS;
3046 : }
3047 :
3048 0 : HcclResult HcclGetInstSizeByNetLayerV2(HcclComm comm, uint32_t netLayer, uint32_t* rankNum)
3049 : {
3050 0 : HCCL_RUN_INFO("Entry-HcclGetInstSizeByNetLayerV2 V950");
3051 0 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
3052 0 : auto ret = communicator->GetInstSizeByNetLayer(netLayer, rankNum);
3053 0 : if (ret != HCCL_SUCCESS) {
3054 0 : HCCL_ERROR(
3055 : "HcclGetInstSizeByNetLayerV2 get InstSize from communicator failed at netLayer[%u], commId[%s], ret[%d]",
3056 : netLayer, communicator->GetId().c_str(), ret);
3057 0 : return HCCL_E_NOT_FOUND; // 查询失败返回
3058 : }
3059 : /* 关键状态记录 */
3060 0 : HCCL_INFO(
3061 : "HcclGetInstSizeByNetLayerV2 success, netLayer[%u], rankNum[%u], commId[%s]", netLayer, *rankNum,
3062 : communicator->GetId().c_str());
3063 0 : return HCCL_SUCCESS;
3064 : }
3065 :
3066 2 : HcclResult HcclGetInstRanksByNetLayerV2(HcclComm comm, uint32_t netLayer, uint32_t** ranks, uint32_t* rankNum)
3067 : {
3068 6 : HCCL_RUN_INFO("Entry-HcclGetInstRanksByNetLayerV2 V950");
3069 2 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
3070 2 : auto ret = communicator->GetInstRanksByNetLayer(netLayer, ranks, rankNum);
3071 2 : if (ret != HCCL_SUCCESS) {
3072 3 : HCCL_ERROR(
3073 : "HcclGetInstRanksByNetLayerV2 get ranks from communicator failed at netLayer[%u], commId[%s], ret[%d]",
3074 : netLayer, communicator->GetId().c_str(), ret);
3075 1 : return HCCL_E_NOT_FOUND; // 查询失败返回
3076 : }
3077 : /* 关键状态记录 */
3078 3 : HCCL_INFO(
3079 : "HcclGetInstRanksByNetLayerV2 success, netLayer[%u], rankNum[%u], commId[%s]", netLayer, *rankNum,
3080 : communicator->GetId().c_str());
3081 1 : return HCCL_SUCCESS;
3082 : }
3083 :
3084 2 : HcclResult HcclGetInstTopoTypeByNetLayerV2(HcclComm comm, uint32_t netLayer, uint32_t* topoType)
3085 : {
3086 6 : HCCL_RUN_INFO("Entry-HcclGetInstTopoTypeByNetLayer V950");
3087 2 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
3088 2 : auto ret = communicator->GetInstTopoTypeByNetLayer(netLayer, topoType);
3089 2 : if (ret != HCCL_SUCCESS) {
3090 3 : HCCL_ERROR(
3091 : "HcclGetInstTopoTypeByNetLayerV2 get topoType from communicator failed at netLayer[%u], commId[%s], "
3092 : "ret[%d]",
3093 : netLayer, communicator->GetId().c_str(), ret);
3094 1 : return HCCL_E_NOT_FOUND;
3095 : }
3096 : /* 关键状态记录 */
3097 3 : HCCL_INFO(
3098 : "HcclGetInstTopoTypeByNetLayer success, netLayer[%u] topoType[%u] commId[%s]", netLayer, *topoType,
3099 : communicator->GetId().c_str());
3100 1 : return HCCL_SUCCESS;
3101 : }
3102 :
3103 : HcclResult
3104 2 : HcclGetInstSizeListByNetLayerV2(HcclComm comm, uint32_t netLayer, uint32_t** instSizeList, uint32_t* listSize)
3105 : {
3106 6 : HCCL_RUN_INFO("Entry-HcclGetInstSizeListByNetLayer V950");
3107 2 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
3108 2 : auto ret = communicator->GetInstSizeListByNetLayer(netLayer, instSizeList, listSize);
3109 2 : if (ret != HCCL_SUCCESS) {
3110 3 : HCCL_ERROR(
3111 : "HcclGetInstSizeListByNetLayerV2 get netInstance size from communicator failed at netLayer[%u], "
3112 : "commId[%s], ret[%d]",
3113 : netLayer, communicator->GetId().c_str(), ret);
3114 1 : return HCCL_E_NOT_FOUND;
3115 : }
3116 : /* 关键状态记录 */
3117 3 : HCCL_INFO(
3118 : "HcclGetInstSizeListByNetLayer success, netLayer[%u] listSize[%u] commId[%s]", netLayer, *listSize,
3119 : communicator->GetId().c_str());
3120 1 : return HCCL_SUCCESS;
3121 : }
3122 :
3123 2 : HcclResult HcclGetLinksV2(
3124 : HcclComm comm, uint32_t netLayer, uint32_t srcRank, uint32_t dstRank, CommLink** linkList, uint32_t* listSize)
3125 : {
3126 6 : HCCL_RUN_INFO("Entry-HcclGetInstSizeListByNetLayer V950");
3127 2 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
3128 2 : auto ret = communicator->GetLinks(netLayer, srcRank, dstRank, linkList, listSize);
3129 2 : if (ret != HCCL_SUCCESS) {
3130 3 : HCCL_ERROR(
3131 : "HcclGetLinksV2 get links from communicator failed at netLayer[%u], commId[%s], ret[%d]", netLayer,
3132 : communicator->GetId().c_str(), ret);
3133 1 : return HCCL_E_NOT_FOUND;
3134 : }
3135 : /* 关键状态记录 */
3136 3 : HCCL_INFO(
3137 : "HcclGetLinks success, netLayer[%u], srcRank[%u], dstRank[%u], listSize[%u], commId[%s]", netLayer, srcRank,
3138 : dstRank, *listSize, communicator->GetId().c_str());
3139 1 : return HCCL_SUCCESS;
3140 : }
3141 :
3142 2 : HcclResult HcclGetTopoInstsByLayerV2(HcclComm comm, uint32_t netLayer, uint32_t** topoInsts, uint32_t* topoInstNum)
3143 : {
3144 6 : HCCL_RUN_INFO("Entry-HcclGetTopoInstsByLayer V950");
3145 2 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
3146 2 : auto ret = communicator->GetTopoInstsByLayer(netLayer, topoInsts, topoInstNum);
3147 2 : if (ret != HCCL_SUCCESS) {
3148 3 : HCCL_ERROR(
3149 : "HcclGetTopoInstsByLayer get topoInsts from communicator failed at netLayer[%u], commId[%s], ret[%d]",
3150 : netLayer, communicator->GetId().c_str(), ret);
3151 1 : return HCCL_E_NOT_FOUND;
3152 : }
3153 : /* 关键状态记录 */
3154 3 : HCCL_INFO(
3155 : "HcclGetTopoInstsByLayer success, netLayer[%u], topoInstNum[%u], commId[%s]", netLayer, *topoInstNum,
3156 : communicator->GetId().c_str());
3157 1 : return HCCL_SUCCESS;
3158 : }
3159 :
3160 2 : HcclResult HcclGetTopoTypeV2(HcclComm comm, uint32_t netLayer, uint32_t topoInstId, CommTopo* topoType)
3161 : {
3162 6 : HCCL_RUN_INFO("Entry-HcclGetInstSizeListByNetLayer V950");
3163 2 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
3164 2 : auto ret = communicator->GetTopoType(netLayer, topoInstId, topoType);
3165 2 : if (ret != HCCL_SUCCESS) {
3166 3 : HCCL_ERROR(
3167 : "HcclGetTopoType get topoType from communicator failed at netLayer[%u], commId[%s], ret[%d]", netLayer,
3168 : communicator->GetId().c_str(), ret);
3169 1 : return HCCL_E_NOT_FOUND;
3170 : }
3171 : /* 关键状态记录 */
3172 3 : HCCL_INFO(
3173 : "HcclGetInstSizeListByNetLayer success, netLayer[%u] topoType[%u] commId[%s]", netLayer, *topoType,
3174 : communicator->GetId().c_str());
3175 1 : return HCCL_SUCCESS;
3176 : }
3177 :
3178 : HcclResult
3179 2 : HcclGetRanksByTopoInstV2(HcclComm comm, uint32_t netLayer, uint32_t topoInstId, uint32_t** ranks, uint32_t* rankNum)
3180 : {
3181 6 : HCCL_RUN_INFO("Entry-HcclGetRanksByTopoInst V950");
3182 2 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
3183 2 : auto ret = communicator->GetRanksByTopoInst(netLayer, topoInstId, ranks, rankNum);
3184 2 : if (ret != HCCL_SUCCESS) {
3185 3 : HCCL_ERROR(
3186 : "HcclGetTopoInstsByLayer get ranks from communicator failed at netLayer[%u], commId[%s], ret[%d]", netLayer,
3187 : communicator->GetId().c_str(), ret);
3188 1 : return HCCL_E_NOT_FOUND;
3189 : }
3190 : /* 关键状态记录 */
3191 3 : HCCL_INFO("HcclGetRanksByTopoInst success, netLayer[%u] rankNum[%u]", netLayer, *rankNum);
3192 1 : return HCCL_SUCCESS;
3193 : }
3194 :
3195 2 : HcclResult HcclRankGraphGetEndpointNumV2(HcclComm comm, uint32_t layer, uint32_t topoInstId, uint32_t* num)
3196 : {
3197 6 : HCCL_RUN_INFO("Entry-HcclRankGraphGetEndpointNum V950");
3198 2 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
3199 2 : auto ret = communicator->GetEndpointNum(layer, topoInstId, num);
3200 2 : if (ret != HCCL_SUCCESS) {
3201 3 : HCCL_ERROR(
3202 : "HcclRankGraphGetEndpointNum get endpoint num from communicator failed at netLayer[%u] with "
3203 : "topoInstId[%u], commId[%s], ret[%d]",
3204 : layer, topoInstId, communicator->GetId().c_str(), ret);
3205 1 : return HCCL_E_NOT_FOUND;
3206 : }
3207 1 : return HCCL_SUCCESS;
3208 : }
3209 :
3210 3 : HcclResult HcclRankGraphGetEndpointDescV2(
3211 : HcclComm comm, uint32_t layer, uint32_t topoInstId, uint32_t* descNum, EndpointDesc* endpointDesc)
3212 : {
3213 9 : HCCL_RUN_INFO("Entry-HcclRankGraphGetEndpointDesc V950");
3214 3 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
3215 3 : auto ret = communicator->GetEndpointDesc(layer, topoInstId, descNum, endpointDesc);
3216 3 : if (ret != HCCL_SUCCESS) {
3217 3 : HCCL_ERROR(
3218 : "HcclRankGraphGetEndpointDesc get endpoint desc from communicator failed at netLayer[%u], commId[%s], "
3219 : "ret[%d]",
3220 : layer, communicator->GetId().c_str(), ret);
3221 1 : return HCCL_E_NOT_FOUND;
3222 : }
3223 2 : return HCCL_SUCCESS;
3224 : }
3225 :
3226 1 : HcclResult HcclRankGraphGetEndpointInfoV2(
3227 : HcclComm comm, uint32_t rankId, const EndpointDesc* endpointDesc, EndpointAttr endpointAttr, uint32_t infoLen,
3228 : void* info)
3229 : {
3230 3 : HCCL_RUN_INFO("Entry-HcclRankGraphGetEndpointInfo V950");
3231 1 : Hccl::HcclCommunicator* communicator = static_cast<Hccl::HcclCommunicator*>(comm);
3232 1 : auto ret = communicator->GetEndpointInfo(rankId, endpointDesc, endpointAttr, infoLen, info);
3233 1 : if (ret != HCCL_SUCCESS) {
3234 0 : HCCL_ERROR(
3235 : "HcclRankGraphGetEndpointInfo get info from communicator failed with endpointAttr [%d], commId[%s], "
3236 : "ret[%d]",
3237 : static_cast<s32>(endpointAttr), communicator->GetId().c_str(), ret);
3238 0 : return HCCL_E_NOT_FOUND;
3239 : }
3240 1 : return HCCL_SUCCESS;
3241 : }
3242 :
3243 1 : HcclResult HcclCommWorkingDevNicSetV2(const HcclComm comm, uint32_t* ranks, bool* useBackup, uint32_t nRanks)
3244 : {
3245 : (void)comm;
3246 : (void)ranks;
3247 : (void)useBackup;
3248 : (void)nRanks;
3249 3 : HCCL_ERROR("HcclCommWorkingDevNicSetV2 not support V950.");
3250 1 : return HCCL_E_NOT_SUPPORT;
3251 : }
3252 :
3253 : HcclResult
3254 1 : HcclCommSetMemoryRangeV2(const HcclComm comm, void* baseVirPtr, size_t size, size_t alignment, uint64_t flags)
3255 : {
3256 : (void)comm;
3257 : (void)baseVirPtr;
3258 : (void)size;
3259 : (void)alignment;
3260 : (void)flags;
3261 3 : HCCL_ERROR("HcclCommSetMemoryRangeV2 not support V950.");
3262 1 : return HCCL_E_NOT_SUPPORT;
3263 : }
3264 :
3265 1 : HcclResult HcclCommUnsetMemoryRangeV2(const HcclComm comm, void* baseVirPtr)
3266 : {
3267 : (void)comm;
3268 : (void)baseVirPtr;
3269 3 : HCCL_ERROR("HcclCommUnsetMemoryRangeV2 not support V950.");
3270 1 : return HCCL_E_NOT_SUPPORT;
3271 : }
3272 :
3273 1 : HcclResult HcclCommActivateCommMemoryV2(
3274 : const HcclComm comm, void* virPtr, size_t size, size_t offset, void* handle, uint64_t flags)
3275 : {
3276 : (void)comm;
3277 : (void)virPtr;
3278 : (void)size;
3279 : (void)offset;
3280 : (void)handle;
3281 : (void)flags;
3282 3 : HCCL_ERROR("HcclCommActivateCommMemoryV2 not support V950.");
3283 1 : return HCCL_E_NOT_SUPPORT;
3284 : }
3285 :
3286 1 : HcclResult HcclCommDeactivateCommMemoryV2(const HcclComm comm, void* virPtr)
3287 : {
3288 : (void)comm;
3289 : (void)virPtr;
3290 3 : HCCL_ERROR("HcclCommDeactivateCommMemoryV2 not support V950.");
3291 1 : return HCCL_E_NOT_SUPPORT;
3292 : }
3293 :
3294 0 : uint32_t HcclGetCommConfigCapabilityV2() { return static_cast<uint32_t>(HCCL_COMM_CONFIG_RETRY); }
3295 :
3296 : #ifdef __cplusplus
3297 : }
3298 : #endif // __cplusplus
|
