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 "communicator_impl.h"
12 : #include <memory>
13 : #include <op_type.h>
14 : #include <adapter_error_manager_pub.h>
15 : #include "orion_adapter_rts.h"
16 : #include "orion_adapter_hal.h"
17 : #include "hccl_exception.h"
18 : #include "null_ptr_exception.h"
19 : #include "runtime_api_exception.h"
20 : #include "exception_util.h"
21 : #include "hccp_hdc_manager.h"
22 : #include "hccp_peer_manager.h"
23 : #include "rdma_handle_manager.h"
24 : #include "env_config.h"
25 : #include "coll_service_ai_cpu_impl.h"
26 : #include "checkcrc.h"
27 : #include "task_exception_handler.h"
28 : #include "coll_service_device_mode.h"
29 : #include "dlprof_function.h"
30 : #include "kfc.h"
31 : #include "op_params_checker.h"
32 : #include "diff_rank_updater.h"
33 : #include "coll_alg_component_builder.h"
34 : #include "coll_alg_component.h"
35 : #include "hccl_common_v2.h"
36 : #include "tp_manager.h"
37 : #include "hccl_aiv_utils.h"
38 : #include "rts_1ton_cnt_notify.h"
39 : #include "rts_cnt_notify.h"
40 : #include "stream_utils.h"
41 : #include "port.h"
42 : #include "net_instance.h"
43 : #include "ascend_hal_base.h"
44 : #include "acl/acl_rt.h"
45 : #include "types.h"
46 : #include "ccu_jetty_mgr.h"
47 : #include "comm_topo_desc.h"
48 : #include "hostdpu/flush_manager.h"
49 : #include "hostdpu/dpu_kernel_entrance.h"
50 : #include "p2p_enable_manager.h"
51 : #include "adapter_error_manager_pub.h"
52 : #include "ccu_context_all_to_all_v_mesh1d.h"
53 : #include "topo_addr_info.h"
54 :
55 : namespace Hccl {
56 : constexpr u64 HCCL_CCL_COMM_FIXED_CALC_BUFFER_SIZE = (1 * 1024 * 1024); // 指定bufferSize的单位为MB
57 : constexpr u64 HCCL_AIV_OFFLOAD_TAG_BUFFER_SIZE = (4 * 1024 * 1024); // 指定bufferSize的单位为MB
58 : constexpr u64 HCCL_MC2_ON_AICPU_FIXED_CALC_BUFFER_SIZE = 1 * HCCL_CCL_COMM_FIXED_CALC_BUFFER_SIZE; // MC2适配AICPU，额外需要1M
59 : std::atomic<u32> Hccl::CommunicatorImpl::globalIndex(0);
60 : constexpr u64 HCCL_CCL_AIV_TAG_BUFFER_SIZE = 2; // 指定存放aiv tag的大小为2M
61 : constexpr u32 HCCL_CCL_AIV_CLEAR_STEP_MAX = 1000; // aiv tag算子下发时++，大于1000置位
62 : constexpr u32 BASE_BIT = 1; // 用于左移设置二进制数的特定位
63 : constexpr u64 SHARE_HBM_MEMORY_SIZE = (100 * 1024 * 1024);
64 : constexpr u64 DPU_TASKEXCEPTION_MEMORY_SIZE = 10; // DPU TASKEXCEPTION共享内存大小 |stopflag[1]|hcclret[2]|hcclret[2]|
65 : constexpr u64 ALIGN_4K = 4096U;
66 : constexpr const char* DPUTAG = "DPUTAG";
67 : constexpr const char* DPUTASKEXCEPTION = "DPUTASKEXCEPTION";
68 : constexpr u64 INDEPENDENT_OP_BUFFER_SIZE_TIMES = 2; //自定义算子buffer倍数
69 : constexpr uint8_t DEVICE_SIGNAL_SECOND = 2;
70 : constexpr uint8_t DEVICE_SIGNAL_THIRD = 3;
71 : constexpr uint32_t TEMP_DEV_TYPE_DPU = 0; // 临时适配，后续rts接口上库之后使用rts的定义
72 : static std::atomic<u32> g_commNum(0); // 一个进程内创建的通信域数量
73 :
74 :
75 : // 支持零拷贝算子的白名单
76 : std::set<OpType> opWhiteSet = {
77 : OpType::BROADCAST,
78 : OpType::ALLTOALL,
79 : OpType::ALLTOALLV,
80 : OpType::SEND,
81 : OpType::RECV,
82 : OpType::ALLGATHER
83 : };
84 :
85 13 : static void PrintBackTrace(HcclException &e)
86 : {
87 13 : auto backTraces = e.GetBackTraceStrings();
88 13 : std::for_each(backTraces.begin(), backTraces.end(), [](string item) {
89 585 : HCCL_INFO(item.c_str());
90 195 : });
91 13 : }
92 :
93 46 : HcclResult CommunicatorImpl::Init(const CommParams &commParams, const std::string &ranktableM,
94 : const HcclCommConfig &config)
95 : {
96 46 : if (!initFlag) {
97 45 : initFlag = true;
98 : try {
99 45 : InitCommonData(commParams, config);
100 45 : InitHccpHdc(); // tsdOpen + rainit
101 45 : InitRankGraph(ranktableM);
102 38 : CHK_RET(InitCommResource(commParams));
103 7 : } catch (HcclException &e) {
104 21 : HCCL_ERROR(e.what());
105 7 : PrintBackTrace(e);
106 7 : return e.GetErrorCode();
107 7 : } catch (exception &e) {
108 0 : HCCL_ERROR(e.what());
109 0 : return HcclResult::HCCL_E_INTERNAL;
110 0 : } catch (...) {
111 0 : HCCL_ERROR("Unknown error occurs!");
112 0 : return HcclResult::HCCL_E_INTERNAL;
113 0 : }
114 38 : return HcclResult::HCCL_SUCCESS;
115 : }
116 3 : HCCL_ERROR("Repeated calling init method!");
117 1 : return HcclResult::HCCL_E_INTERNAL;
118 : }
119 :
120 38 : HcclResult CommunicatorImpl::InitCommResource(const CommParams &commParams)
121 : {
122 38 : HrtSetDevice(devLogicId);
123 38 : if (IsNeedDpu()) {
124 5 : InitHccpPeer();
125 : }
126 38 : AppendLocalDieIdForLinks();
127 38 : InitCcuSuperFastLoad();
128 38 : InitNotifyManager();
129 38 : InitStreamManager();
130 38 : InitPreResource();
131 38 : InitSocketManager();
132 38 : InitRmaConnManager();
133 38 : InitDataBufferManager();
134 38 : InitMemTransportManager();
135 38 : InitHostDeviceSyncNotifyManager();
136 38 : InitUbMemoryTransportMgr();
137 38 : CollAlgComponentInit(); // 初始化算法组件
138 38 : RegisterAicpuKernel();
139 38 : InitCollService();
140 38 : InitTraceManager();
141 38 : DlProfFunction::GetInstance().DlProfFunctionInit();
142 38 : InitMirrorTaskManager();
143 38 : CHK_RET(InitProfilingReporter());
144 38 : InitTaskExceptionHandler();
145 38 : InitHDCommunicate();
146 38 : notifyTimeoutCfg.Init();
147 38 : SetCommStatus(CommStatus::COMM_READY);
148 38 : SnapShotParser::GetInstance().SerializeCommonInfo(commParams, config, std::move(ranktableInfo), topoInfo, staticBinaryInfo);
149 38 : InitOneSidedService();
150 38 : RegisterKernel();
151 38 : InitDpuKernel();
152 38 : return HCCL_SUCCESS;
153 : }
154 :
155 39 : void CommunicatorImpl::InitDpuKernel() {
156 39 : std::unordered_set<IpAddress> hostIps = GetHostIpFromRankGraph();
157 39 : if (hostIps.empty()) {
158 39 : return;
159 : }
160 0 : for (auto ip: hostIps) {
161 0 : FlushManager::GetInstance().initFlushHandle(ip, devPhyId);
162 : }
163 0 : HCCL_INFO("[InitDpuKernel]all FlushHandle init success.");
164 : /* kernel Launch */
165 0 : CHK_RET_THROW(RuntimeApiException, "InitAndLaunchDpuKernel Failed", InitAndLaunchDpuKernel());
166 0 : CHK_RET_THROW(RuntimeApiException, "InitAndLaunchAicpuKernel Failed", InitAndLaunchAicpuKernel()); // 下aicpukernel,将taskexception共享内存保存到aicpu全局map中
167 39 : }
168 :
169 39 : std::unordered_set<IpAddress> CommunicatorImpl::GetHostIpFromRankGraph()
170 : {
171 117 : HCCL_INFO("[GetHostIpFromRankGraph]Start get host ip.");
172 39 : std::unordered_set<IpAddress> ips;
173 39 : if (rankGraph->GetPeer(myRank) == nullptr) {
174 0 : HCCL_ERROR("[GetHostIpFromRankGraph] rankGraph peer is null!");
175 0 : return ips;
176 : }
177 39 : std::vector<std::shared_ptr<NetInstance::ConnInterface>> interfaces = rankGraph->GetPeer(myRank)->GetIfaces();
178 64 : for (auto interface : interfaces) {
179 : // 找到所有在host上和LinkProtocol有rdma的ip进行注册
180 25 : if (interface->GetPos() == AddrPosition::HOST && interface->GetLinkProtocols().count(LinkProtocol::ROCE) != 0) {
181 0 : IpAddress ip = interface->GetAddr();
182 0 : ips.insert(ip);
183 : }
184 25 : }
185 117 : HCCL_INFO("[GetHostIpFromRankGraph] Successfully completed: GetHostIp finished.");
186 39 : return ips;
187 39 : }
188 :
189 0 : HcclResult CommunicatorImpl::Init(const CommParams &commParams, const RankTableInfo &ranktable,
190 : const HcclCommConfig &config)
191 : {
192 0 : if (!initFlag) {
193 0 : initFlag = true;
194 : try {
195 0 : InitCommonData(commParams, config);
196 0 : InitHccpHdc(); // tsdOpen + rainit
197 0 : InitRankGraph(ranktable);
198 0 : CHK_RET(InitCommResource(commParams));
199 0 : } catch (HcclException &e) {
200 0 : HCCL_ERROR(e.what());
201 0 : PrintBackTrace(e);
202 0 : return e.GetErrorCode();
203 0 : } catch (exception &e) {
204 0 : HCCL_ERROR(e.what());
205 0 : return HcclResult::HCCL_E_INTERNAL;
206 0 : } catch (...) {
207 0 : HCCL_ERROR("Unknown error occurs!");
208 0 : return HcclResult::HCCL_E_INTERNAL;
209 0 : }
210 0 : return HcclResult::HCCL_SUCCESS;
211 : }
212 0 : HCCL_ERROR("Repeated calling init method!");
213 0 : return HcclResult::HCCL_E_INTERNAL;
214 : }
215 :
216 1 : HcclResult CommunicatorImpl::Init(const CommParams &commParams, std::unique_ptr<RankGraph> &inputRankGraph, DevId inputDevLogicId)
217 : {
218 1 : if (!initFlag) {
219 1 : initFlag = true;
220 : try {
221 1 : HrtSetDevice(inputDevLogicId);
222 1 : InitCommonData(commParams);
223 1 : InitRankGraph(inputRankGraph);
224 1 : HrtSetDevice(devLogicId);
225 1 : if (IsNeedDpu()) {
226 0 : InitHccpPeer();
227 : }
228 1 : InitHccpHdc();
229 1 : AppendLocalDieIdForLinks();
230 1 : InitCcuSuperFastLoad();
231 1 : InitNotifyManager();
232 1 : InitStreamManager();
233 1 : InitSocketManager();
234 1 : InitRmaConnManager();
235 1 : InitDataBufferManager();
236 1 : InitMemTransportManager();
237 1 : InitHostDeviceSyncNotifyManager();
238 1 : InitUbMemoryTransportMgr();
239 1 : CollAlgComponentInit();
240 1 : RegisterAicpuKernel();
241 1 : InitCollService();
242 1 : InitTraceManager();
243 1 : InitHDCommunicate();
244 1 : InitMirrorTaskManager();
245 1 : CHK_RET(InitProfilingReporter());
246 1 : InitTaskExceptionHandler();
247 1 : RegisterKernel();
248 1 : InitDpuKernel();
249 1 : SetCommStatus(CommStatus::COMM_READY);
250 0 : } catch (HcclException &e) {
251 0 : HCCL_ERROR(e.what());
252 0 : PrintBackTrace(e);
253 0 : return e.GetErrorCode();
254 0 : } catch (exception &e) {
255 0 : HCCL_ERROR(e.what());
256 0 : return HcclResult::HCCL_E_INTERNAL;
257 0 : } catch (...) {
258 0 : HCCL_ERROR("Unknown error occurs!");
259 0 : return HcclResult::HCCL_E_INTERNAL;
260 0 : }
261 1 : return HcclResult::HCCL_SUCCESS;
262 : }
263 0 : HCCL_ERROR("Repeated calling init method!");
264 0 : return HcclResult::HCCL_E_INTERNAL;
265 : }
266 :
267 1 : HcclResult CommunicatorImpl::Init(const CommParams &commParams, std::unique_ptr<RankGraph> &inputRankGraph,
268 : HcclCommConfig &subConfig, DevId inputDevLogicId)
269 : {
270 1 : if (!initFlag) {
271 0 : initFlag = true;
272 0 : TRY_CATCH_RETURN(
273 : HrtSetDevice(inputDevLogicId);
274 : InitCommonData(commParams, subConfig);
275 : InitHccpHdc();
276 : InitCcuSuperFastLoad();
277 : InitNotifyManager();
278 : InitStreamManager();
279 : InitSocketManager();
280 : InitRmaConnManager();
281 : InitDataBufferManager();
282 : InitMemTransportManager();
283 : InitHostDeviceSyncNotifyManager();
284 : InitTraceManager();
285 : InitHDCommunicate();
286 : notifyTimeoutCfg.Init();
287 : InitRankGraph(inputRankGraph);
288 : if (IsNeedDpu()) {
289 : InitHccpPeer();
290 : }
291 : AppendLocalDieIdForLinks();
292 : InitUbMemoryTransportMgr();
293 : CollAlgComponentInit();
294 : RegisterAicpuKernel();
295 : InitCollService();
296 : DlProfFunction::GetInstance().DlProfFunctionInit();
297 : InitMirrorTaskManager();
298 : CHK_RET(InitProfilingReporter());
299 : InitTaskExceptionHandler();
300 : RegisterKernel();
301 : InitDpuKernel();
302 : SetCommStatus(CommStatus::COMM_READY);
303 : SnapShotParser::GetInstance().SerializeSubCommInfo(commParams, subConfig, rankIdsVec, staticBinaryInfo);
304 : );
305 0 : return HcclResult::HCCL_SUCCESS;
306 : } else {
307 3 : HCCL_ERROR("Repeated calling init method!");
308 1 : return HcclResult::HCCL_E_INTERNAL;
309 : }
310 : }
311 :
312 2 : HcclResult CommunicatorImpl::CreateSubComm(const CommParams &subCommParams, const std::vector<u32> &rankIds,
313 : CommunicatorImpl *subCommImpl)
314 : {
315 130 : TRY_CATCH_RETURN(
316 : if (initFlag) {
317 : // 创建子虚拟拓扑
318 : std::unique_ptr<RankGraph> subRankGraph = rankGraph->CreateSubRankGraph(rankIds);
319 : // 初始化子通信域
320 : CHK_RET(subCommImpl->Init(subCommParams, subRankGraph, devLogicId));
321 : auto rankIpPortMap = GetSocketManager().GetSubCommDeviceServerListenPortMap(rankIds);
322 : RankIpPortMapPtr rankIpPortMapPtr = std::make_shared<decltype(rankIpPortMap)>(std::move(rankIpPortMap));
323 : CHK_RET(subCommImpl->SetRankIpPortMap(rankIpPortMapPtr));
324 : subCommImpl->GetSocketManager().SetDeviceServerListenPortMap(*rankIpPortMapPtr);
325 : return HcclResult::HCCL_SUCCESS;
326 : } else {
327 : std::string msg = StringFormat("CreateSubComm fail, communicator has not been initialized, please check.");
328 : THROW<InternalException>(msg);
329 : }
330 : );
331 : HCCL_ERROR("CreateSubComm fail !");
332 : return HcclResult::HCCL_E_INTERNAL;
333 : }
334 :
335 2 : HcclResult CommunicatorImpl::CreateSubComm(const CommParams &subCommParams, const std::vector<u32> &rankIds,
336 : CommunicatorImpl *subCommImpl, HcclCommConfig &subConfig)
337 : {
338 130 : TRY_CATCH_RETURN(
339 : if (initFlag) {
340 : // 创建子虚拟拓扑
341 : std::unique_ptr<RankGraph> subRankGraph = rankGraph->CreateSubRankGraph(rankIds);
342 : subCommImpl->rankIdsVec = rankIds;
343 : HCCL_INFO("[%s]rankIds size[%u], rankIdsVec size[%u]", __func__, rankIds.size(), subCommImpl->rankIdsVec.size());
344 : // 初始化子通信域
345 : CHK_RET(subCommImpl->Init(subCommParams, subRankGraph, subConfig, devLogicId));
346 : auto rankIpPortMap = GetSocketManager().GetSubCommDeviceServerListenPortMap(rankIds);
347 : RankIpPortMapPtr rankIpPortMapPtr = std::make_shared<decltype(rankIpPortMap)>(std::move(rankIpPortMap));
348 : CHK_RET(subCommImpl->SetRankIpPortMap(rankIpPortMapPtr));
349 : subCommImpl->GetSocketManager().SetDeviceServerListenPortMap(*rankIpPortMapPtr);
350 : return HcclResult::HCCL_SUCCESS;
351 : } else {
352 : std::string msg = StringFormat("CreateSubComm fail, communicator has not been initialized, please check.");
353 : THROW<InternalException>(msg);
354 : }
355 : );
356 : HCCL_ERROR("CreateSubComm fail !");
357 : return HcclResult::HCCL_E_INTERNAL;
358 : }
359 :
360 6 : void CommunicatorImpl::RefreshSubmittedOpcnt()
361 : {
362 6 : if (currentCollOperator->opType == OpType::SEND || currentCollOperator->opType == OpType::RECV) {
363 0 : sendRecvIndex++;
364 0 : submittedOpCnt = sendRecvIndex;
365 : } else {
366 6 : collOpIndex++;
367 6 : submittedOpCnt = collOpIndex;
368 : }
369 18 : HCCL_INFO("[%s] end, opType[%s], submittedOpCnt[%u], sendRecvIndex[%u], collOpIndex[%u].", __func__,
370 : currentCollOperator->opType.Describe().c_str(), submittedOpCnt, sendRecvIndex, collOpIndex);
371 6 : }
372 :
373 7 : void CommunicatorImpl::SingleRankProc(const CollOpParams &opParams, void *stream) const
374 : {
375 13 : if (opParams.opType == Hccl::OpType::BATCHSENDRECV || opParams.opType == Hccl::OpType::SEND
376 13 : || opParams.opType == Hccl::OpType::RECV) {
377 6 : HCCL_WARNING("[CommunicatorImpl][%s] ranksize == 1 is not support BATCHSENDRECV SEND RECV", __func__);
378 2 : return;
379 : }
380 5 : if (opParams.sendBuf == opParams.recvBuf) {
381 6 : HCCL_WARNING("[CommunicatorImpl][%s] sendBuf == recvBuf, return success", __func__);
382 2 : return;
383 : }
384 3 : u64 len{0};
385 3 : if (opParams.opType == Hccl::OpType::ALLTOALL) {
386 1 : len = DataTypeSizeGet(opParams.all2AllDataDes.sendType) * opParams.all2AllDataDes.sendCount;
387 2 : } else if (opParams.opType == Hccl::OpType::ALLTOALLV) {
388 1 : len = DataTypeSizeGet(opParams.all2AllVDataDes.sendType) * *(static_cast<const u64 *>(opParams.all2AllVDataDes.sendCounts));
389 1 : } else if (opParams.opType == Hccl::OpType::ALLTOALLVC) {
390 0 : len = DataTypeSizeGet(opParams.all2AllVCDataDes.sendType) * *(static_cast<const u64 *>(opParams.all2AllVCDataDes.sendCountMatrix));
391 : } else {
392 1 : len = DataTypeSizeGet(opParams.dataType) * opParams.count;
393 : }
394 :
395 9 : HCCL_INFO("[CommunicatorImpl][%s] sendBuf[%p], recvBuf[%p], len[%llu].", __func__, opParams.sendBuf, opParams.recvBuf, len);
396 3 : if (len > 0) {
397 3 : HrtMemAsyncCopy(opParams.recvBuf, len, opParams.sendBuf, len, ACL_MEMCPY_DEVICE_TO_DEVICE, stream);
398 : }
399 : }
400 :
401 17 : bool CommunicatorImpl::TryFastCcuLaunch(const CollOpParams &opParams, aclrtStream const stream)
402 : {
403 17 : InitCcuSuperFastLoad(); // 存在profiling开关在多次下发算子时动态变化的场景，每次下发流程中都需要更新开关
404 25 : superFasterLoad = (opParams.opType == OpType::ALLREDUCE || opParams.opType == OpType::ALLGATHER ||
405 10 : opParams.opType == OpType::REDUCESCATTER || opParams.opType == OpType::BROADCAST ||
406 7 : opParams.opType == OpType::ALLTOALL || opParams.opType == OpType::REDUCE ||
407 23 : opParams.opType == OpType::SCATTER || opParams.opType == OpType::ALLTOALLV
408 : );
409 20 : bool canUpdate = superFasterLoad && (commExecuteConfig.accState == AcceleratorState::CCU_MS ||
410 3 : commExecuteConfig.accState == AcceleratorState::CCU_SCHED);
411 17 : if (OpType::ALLTOALL == opParams.opType) {
412 1 : ccuParamsMappingKey = {static_cast<u32>(opParams.reduceOp), static_cast<u32>(opParams.all2AllDataDes.sendType), static_cast<u32>(opParams.all2AllDataDes.sendCount)};
413 16 : } else if (OpType::ALLTOALLV == opParams.opType) {
414 0 : ccuParamsMappingKey = {static_cast<u32>(opParams.reduceOp), static_cast<u32>(opParams.all2AllVDataDes.sendType), 0};
415 16 : } else if (OpType::BROADCAST == opParams.opType || OpType::SCATTER == opParams.opType) {
416 0 : ccuParamsMappingKey = {static_cast<u32>(opParams.root), static_cast<u32>(opParams.dataType), static_cast<u32>(opParams.count)};
417 : } else {
418 16 : ccuParamsMappingKey = {static_cast<u32>(opParams.reduceOp), static_cast<u32>(opParams.dataType), static_cast<u32>(opParams.count)};
419 : }
420 17 : auto &ccuParamsMapping = colCcuParamMapping[opParams.opType];
421 17 : auto ccuParamsMappingKeyIter = ccuParamsMapping.find(ccuParamsMappingKey);
422 17 : bool isCCUChangeModel = canUpdate && ccuParamsMappingKeyIter != ccuParamsMapping.end();
423 17 : if (!isCCUChangeModel) {
424 12 : return false;
425 : }
426 5 : CachedCCUParams ¶ms = ccuParamsMappingKeyIter->second;
427 :
428 5 : if (opParams.opType == OpType::ALLTOALLV && params.insType != CcuInstType::CCU_ALLTOALLV_MESH_1D_DIRECT) {
429 0 : return false;
430 : }
431 5 : if (enableProfilingEnv) {
432 1 : uint64_t beginTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
433 1 : UpdateProfStat();
434 1 : auto dfxOpInfo = std::make_shared<DfxOpInfo>();
435 1 : CovertToCurrentCollOperator(id, opParams, OpMode::OPBASE);
436 1 : dfxOpInfo->op_ = *GetCurrentCollOperator();
437 1 : dfxOpInfo->tag_ = dfxOpInfo->op_.opTag;
438 1 : dfxOpInfo->algType_ = GetCurAlgName().c_str();
439 1 : dfxOpInfo->commIndex_ = GetIdIndex();
440 1 : dfxOpInfo->comm_ = this;
441 1 : dfxOpInfo->beginTime_ = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
442 1 : dfxOpInfo->commId_ = id;
443 1 : dfxOpInfo->opIndex_ = opIndex;
444 1 : GetMirrorTaskManager().SetCurrDfxOpInfo(dfxOpInfo);
445 1 : ExecuteFastCcuLaunch(opParams, stream, params);
446 1 : ReportProfInfo(beginTime, opParams.staticShape, true);
447 1 : } else {
448 4 : ExecuteFastCcuLaunch(opParams, stream, params);
449 : }
450 5 : return true;
451 : }
452 :
453 9 : static void FastCcuLaunchSaveDfxTaskInfo(const CommunicatorImpl &comm, const TaskParam &taskParam, bool isMaster,
454 : const u32 remoteRankId = INVALID_VALUE_RANKID)
455 : {
456 : u32 taskId;
457 : u32 streamId;
458 9 : HrtGetTaskIdAndStreamID(taskId, streamId);
459 :
460 : std::unique_ptr<TaskInfo> taskInfo = std::make_unique<TaskInfo>(streamId, taskId, remoteRankId, taskParam,
461 9 : comm.GetMirrorTaskManager().GetCurrDfxOpInfo(), isMaster);
462 :
463 27 : HCCL_INFO("Begin to AddTaskInfo: streamId[%lu], taskId[%lu], remoteRankId[%u].", streamId, taskId, remoteRankId);
464 9 : comm.GetMirrorTaskManager().AddTaskInfo(std::move(taskInfo));
465 9 : }
466 :
467 0 : void CommunicatorImpl::FillAllToAllVArgs(const CollOpParams &opParams, rtCcuTaskInfo_t *&ccuParams) const
468 : {
469 0 : std::vector<uint64_t> args;
470 0 : CcuContextAllToAllVMesh1D::RefreshArgs(opParams, rankSize, args, myRank);
471 0 : rtCcuTaskInfo_t *currCcuParam = ccuParams;
472 0 : for (u32 i = 0; i < args.size(); i++) {
473 : // skip token info
474 0 : if (i == 2) {
475 0 : continue;
476 : }
477 0 : currCcuParam->args[i % RT_CCU_SQE_ARGS_LEN] = args[i];
478 0 : if ((i + 1) % RT_CCU_SQE_ARGS_LEN == 0) {
479 0 : currCcuParam += 1;
480 : }
481 : }
482 0 : }
483 :
484 3 : void CommunicatorImpl::ExecuteFastCcuLaunch(const CollOpParams &opParams, aclrtStream const stream, CachedCCUParams ¶ms)
485 : {
486 : static thread_local int slaveIndex = 0;
487 : static thread_local u32 mStreamId = 0;
488 : static thread_local u32 value = 0;
489 : static thread_local Rts1ToNCntNotify *cntNotify1ToN = nullptr;
490 3 : static thread_local u32 timeout = notifyTimeoutCfg.GetNotifyTimeout();
491 :
492 3 : rtCcuTaskInfo_t *&ccuParams = params.ccuParams;
493 :
494 3 : if (params.insType == CcuInstType::CCU_ALLTOALLV_MESH_1D_DIRECT) {
495 0 : FillAllToAllVArgs(opParams, ccuParams);
496 6 : } else if (params.insType == CcuInstType::CCU_ALLTOALL_MESH_1D_2DIE ||
497 6 : params.insType == CcuInstType::CCU_ALLGATHER_MESH_1D_2DIE ||
498 3 : params.insType == CcuInstType::CCU_REDUCE_SCATTER_MESH_1D_2DIE) {
499 0 : for (std::size_t i = 0; i < params.totalCounts; ++i) {
500 0 : (void)memcpy_s(&ccuParams[i].args[0], sizeof(ccuParams[i].args[0]), &opParams.sendBuf,
501 : sizeof(ccuParams[i].args[0]));
502 0 : (void)memcpy_s(&ccuParams[i].args[1], sizeof(ccuParams[i].args[1]), &opParams.recvBuf,
503 : sizeof(ccuParams[i].args[1]));
504 : }
505 : } else {
506 3 : (void)memcpy_s(&ccuParams[0].args[0], sizeof(ccuParams[0].args[0]), &opParams.sendBuf,
507 : sizeof(ccuParams[0].args[0]));
508 3 : (void)memcpy_s(&ccuParams[0].args[1], sizeof(ccuParams[0].args[1]), &opParams.recvBuf,
509 : sizeof(ccuParams[0].args[1]));
510 : }
511 :
512 3 : auto vector_zero_count = params.count[0];
513 3 : auto &opbaseStream = GetStreamManager().opbase;
514 3 : auto mStream = params.isSlave ? opbaseStream->GetSlave(slaveIndex)->GetPtr() : stream;
515 3 : u32 streamNum = params.count.size();
516 3 : if (streamNum > 1) {
517 3 : timeout = notifyTimeoutCfg.GetNotifyTimeout();
518 3 : mStreamId = params.isSlave ? opbaseStream->GetSlave(slaveIndex++)->GetId() : HrtGetStreamId(mStream);
519 3 : cntNotify1ToN = GetCcuStreamSyncNotifyManager().GetRts1ToNCntNotify(mStreamId);
520 : // launch LocalPostTo on stream
521 3 : value = 0;
522 9 : for (u32 i = 0; i < streamNum - 1; ++i) {
523 6 : value |= BASE_BIT << i;
524 : }
525 3 : cntNotify1ToN->PostValue(value, mStream);
526 : }
527 3 : if (taskExceptionEnv || enableProfilingEnv) {
528 3 : params.taskParams[0].beginTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
529 3 : SuperFastLoad(ccuParams, mStream, vector_zero_count);
530 3 : params.taskParams[0].endTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
531 3 : FastCcuLaunchSaveDfxTaskInfo(*this, params.taskParams[0], (!params.isSlave));
532 : } else {
533 0 : SuperFastLoad(ccuParams, mStream, vector_zero_count);
534 : }
535 :
536 3 : if (streamNum > 1) {
537 3 : RtsCntNotify *cntNotifyNTo1 = GetCcuStreamSyncNotifyManager().GetRtsNTo1CntNotify(mStreamId);
538 3 : opbaseStream->RegisterMaster(std::make_unique<Stream>(stream));
539 : // launch LocalWaitFrom on stream
540 3 : cntNotifyNTo1->WaitValue(value, timeout, mStream);
541 9 : for (std::size_t i = 0, len = streamNum - 1; i < len; ++i) {
542 6 : u32 bitValue = BASE_BIT << i;
543 6 : auto slave = opbaseStream->GetSlave(slaveIndex++);
544 6 : auto master = opbaseStream->GetMaster();
545 6 : GetStreamManager().CaptureSlaveStream(master, slave);
546 6 : cntNotify1ToN->WaitBits(bitValue, timeout, *slave);
547 6 : if (taskExceptionEnv || enableProfilingEnv) {
548 6 : params.taskParams[i + 1].beginTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
549 6 : SuperFastLoad(ccuParams + params.count[i], slave->GetPtr(), params.count[i + 1]);
550 6 : params.taskParams[i + 1].endTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
551 6 : FastCcuLaunchSaveDfxTaskInfo(*this, params.taskParams[i + 1], slave->IsMaster());
552 : }
553 : else{
554 0 : SuperFastLoad(ccuParams + params.count[i], slave->GetPtr(), params.count[i + 1]);
555 : }
556 : // launch localPostTo on extra streams
557 6 : cntNotifyNTo1->PostBits(bitValue, *slave);
558 : }
559 : }
560 3 : if(params.insType == CcuInstType::CCU_REDUCE_SCATTER_MESH_1D_2DIE) {
561 : //硬编码
562 0 : if (taskExceptionEnv || enableProfilingEnv) {
563 0 : TaskParam taskParam{};
564 0 : taskParam.beginTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
565 0 : aclrtReduceKind rtReduceOp = static_cast<aclrtReduceKind>(static_cast<int>(RtReduceOpGet(opParams.reduceOp)));
566 0 : aclDataType rtDataType = static_cast<aclDataType>(static_cast<int>(RtDataTypeGet(opParams.dataType)));
567 0 : constexpr std::size_t myScratchPlace = 4;
568 0 : const u32 scratchSize = ccuParams[0].args[myScratchPlace];
569 0 : auto src = reinterpret_cast<void *>(ccuParams[0].args[3]);
570 0 : auto dst = reinterpret_cast<void *>(ccuParams[0].args[1]);
571 0 : HrtReduceAsync(dst, scratchSize, src, scratchSize, rtReduceOp, rtDataType, stream);
572 0 : taskParam.taskType = TaskParamType::TASK_REDUCE_INLINE;
573 0 : taskParam.endTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
574 0 : taskParam.taskPara.Reduce.src = src;
575 0 : taskParam.taskPara.Reduce.dst = dst;
576 0 : taskParam.taskPara.Reduce.size = scratchSize;
577 0 : taskParam.taskPara.Reduce.notifyID = INVALID_VALUE_NOTIFYID;
578 0 : taskParam.taskPara.Reduce.linkType = DfxLinkType::ONCHIP;
579 0 : taskParam.taskPara.Reduce.dataType = DataTypeToHcclDataType(opParams.dataType);
580 0 : taskParam.taskPara.Reduce.reduceOp = ReduceOpToHcclReduceOp(opParams.reduceOp);
581 0 : FastCcuLaunchSaveDfxTaskInfo(*this, taskParam, true, GetMyRank()); // stream为主流
582 0 : } else {
583 0 : aclrtReduceKind rtReduceOp = static_cast<aclrtReduceKind>(static_cast<int>(RtReduceOpGet(opParams.reduceOp)));
584 0 : aclDataType rtDataType = static_cast<aclDataType>(static_cast<int>(RtDataTypeGet(opParams.dataType)));
585 0 : constexpr std::size_t myScratchPlace = 4;
586 0 : const u32 scratchSize = ccuParams[0].args[myScratchPlace];
587 0 : auto src = reinterpret_cast<void *>(ccuParams[0].args[3]);
588 0 : auto dst = reinterpret_cast<void *>(ccuParams[0].args[1]);
589 0 : HrtReduceAsync(dst, scratchSize, src, scratchSize, rtReduceOp, rtDataType, stream);
590 : }
591 : }
592 :
593 3 : slaveIndex = 0;
594 3 : collOpIndex++;
595 3 : submittedOpCnt = collOpIndex;
596 3 : opBaseOpIndex++;
597 3 : opIndex++;
598 3 : SetCommStatus(CommStatus::COMM_READY);
599 3 : }
600 :
601 5 : HcclResult CommunicatorImpl::SetAivControledCoreNum(bool isAiv)
602 : {
603 5 : if (isAiv) {
604 0 : u32 numBlocksLimit = MAX_NUM_BLOCKS;
605 0 : aclError acl_ret = aclrtGetResInCurrentThread(ACL_RT_DEV_RES_VECTOR_CORE, &numBlocksLimit);
606 0 : CHK_PRT_RET(acl_ret != ACL_SUCCESS,
607 : HCCL_ERROR("[CommunicatorImpl::SetAivControledCoreNum] aclrtGetResInCurrentThread failed, ret=[%d]", acl_ret),
608 : HCCL_E_PARA);
609 0 : CHK_PRT_RET(numBlocksLimit < 1,
610 : HCCL_ERROR("[CommunicatorImpl::SetAivControledCoreNum] block num less than 1, block num[%u]", numBlocksLimit),
611 : HCCL_E_PARA);
612 0 : currentCollOperator->numBlocksLimit = numBlocksLimit;
613 0 : HCCL_INFO("[CommunicatorImpl::SetAivControledCoreNum] Aiv core limit is [%u].", numBlocksLimit);
614 : }
615 5 : return HCCL_SUCCESS;
616 : }
617 :
618 17 : static HcclResult MatchAclgraph(const rtStream_t stream, bool &isCapture)
619 : {
620 17 : rtModel_t rtModel = nullptr;
621 17 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
622 17 : return HCCL_SUCCESS;
623 : }
624 :
625 0 : HcclResult CommunicatorImpl::OffloadResourcePre(std::string &opTag, const CollOpParams &opParams)
626 : {
627 0 : CollOffloadOpResReq resReq;
628 0 : auto dataSize = opParams.count * DataTypeSizeGet(opParams.dataType);
629 0 : auto dataType = DataTypeToHcclDataType(opParams.dataType);
630 0 : CHK_RET(CalcCollOffloadOpRes(opParams.opType, dataSize, dataType, resReq));
631 :
632 : // 设定workspace内存资源
633 0 : std::vector<rtStream_t> slaveStreams;
634 0 : slaveStreams.resize(resReq.requiredSubQueNum);
635 0 : for (u64 i = 0; i < resReq.requiredSubQueNum; ++i) {
636 0 : slaveStreams[i] = static_cast<rtStream_t>(std::make_unique<Stream>(true, false).get());
637 : }
638 0 : CHK_RET(SetCollOffloadSlaveStreams(opTag, slaveStreams));
639 0 : CHK_RET(SetCollOffloadScratchBuf(opTag, reinterpret_cast<void *>(GetCclBuffer()->GetAddr()),
640 : GetCclBuffer()->GetSize()));
641 0 : return HCCL_SUCCESS;
642 0 : }
643 :
644 21 : HcclResult CommunicatorImpl::LoadOpbasedCollOp(const CollOpParams &opParams, void *stream)
645 : {
646 : try {
647 21 : isLoadOp = true;
648 40 : CHK_RET(CheckCommStatus());
649 : // 等待通信域状态为Ready，执行算子下发
650 18 : WaitReady();
651 18 : SnapShotParser::GetInstance().SetIsNeedLoadOp(false);
652 18 : if (rankSize == 1) {
653 15 : HCCL_WARNING("[CommunicatorImpl][%s] ranksize == 1, enter SingleRankProc", __func__);
654 5 : SingleRankProc(opParams, stream);
655 5 : return HcclResult::HCCL_SUCCESS;
656 : }
657 : // 判断是否为aclgraph
658 13 : bool isCapture = false; // isCapture为true表示aclgraph
659 13 : CHK_RET(MatchAclgraph(stream, isCapture));
660 13 : if (!isCapture && TryFastCcuLaunch(opParams, stream)) { // 若是aclgraph则不走快速下发
661 3 : return HcclResult::HCCL_SUCCESS;
662 : }
663 10 : curOpParams = opParams;
664 10 : CovertToCurrentCollOperator(id, opParams, OpMode::OPBASE);
665 8 : opExecuteConfig = commExecuteConfig;
666 8 : ExecAlgSelect(opParams, OpMode::OPBASE); // 根据配置选择对应的collService
667 7 : if (dynamic_cast<CollServiceDefaultImpl *>(collService) != nullptr) {
668 3 : HCCL_ERROR("Opbase mode is not supported in expanding on the host in 950");
669 1 : return HcclResult::HCCL_E_NOT_SUPPORT;
670 : }
671 6 : bool isAiv = (opExecuteConfig.accState == AcceleratorState::AIV || opExecuteConfig.accState == AcceleratorState::AIV_ONLY);
672 6 : SetCommStatus(CommStatus::COMM_READY);
673 9 : CHK_RET(OpParamsChecker::CheckOpDataTypeOpbase(opParams, GetOpCcuFeatureFlag(), GetOpAiCpuTSFeatureFlag(), isAiv));
674 :
675 : // AICPU aclgraph场景传入的stream被capture且算子时支持零拷贝算法的,会切换到图模式
676 5 : if (opExecuteConfig.accState == AcceleratorState::AICPU_TS && isCapture && (opWhiteSet.find(opParams.opType) != opWhiteSet.end())) {
677 0 : std::string tag = opParams.opTag + "_" + std::to_string(tagResourceIndex_++);
678 0 : OffloadResourcePre(tag, opParams);
679 0 : HCCL_INFO("[CommunicatorImpl][%s]current op support zero copy in aicpu aclgraph, change to offload", __func__);
680 0 : return LoadOffloadCollOp(tag, opParams, stream);
681 0 : }
682 5 : CHK_RET(SetAivControledCoreNum(isAiv));
683 :
684 : // 避免transport建链前，通讯域被摧毁
685 5 : SetCommStatus(CommStatus::COMM_INUSE);
686 5 : if (opParams.sendBuf != nullptr) {
687 0 : PrintMemoryAttr(opParams.sendBuf);
688 : }
689 5 : if (opParams.recvBuf != nullptr) {
690 0 : PrintMemoryAttr(opParams.recvBuf);
691 : }
692 5 : uint64_t beginTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
693 :
694 : // 更新开关状态
695 5 : UpdateProfStat();
696 5 : collService->LoadWithOpBasedMode(*currentCollOperator, std::make_unique<Stream>(stream));
697 5 : if (++aivTag > HCCL_CCL_AIV_CLEAR_STEP_MAX) {
698 0 : aivTag = 1;
699 : }
700 : // ReportProfInfok:opinfo, allTaskInfo
701 5 : bool cachedReq = opParams.staticShape || isCapture;
702 5 : ReportProfInfo(beginTime, cachedReq, true);
703 5 : RefreshSubmittedOpcnt();
704 5 : opBaseOpIndex++;
705 5 : opIndex++;
706 5 : SetCommStatus(CommStatus::COMM_READY);
707 3 : } catch (HcclException &e) {
708 1 : SetCommStatus(CommStatus::COMM_READY);
709 3 : HCCL_ERROR(e.what());
710 1 : PrintBackTrace(e);
711 1 : u32 idxHcclException = GetSubmittedOpCnt();
712 3 : HCCL_ERROR("SubmittedOpCnt: %u, OperatorParams: %s", idxHcclException, opParams.Describe().c_str());
713 1 : return e.GetErrorCode();
714 2 : } catch (exception &e) {
715 1 : SetCommStatus(CommStatus::COMM_READY);
716 3 : HCCL_ERROR(e.what());
717 1 : u32 idxException = GetSubmittedOpCnt();
718 3 : HCCL_ERROR("SubmittedOpCnt: %u, OperatorParams: %s", idxException, opParams.Describe().c_str());
719 1 : return HcclResult::HCCL_E_INTERNAL;
720 2 : } catch (...) {
721 1 : SetCommStatus(CommStatus::COMM_READY);
722 1 : u32 idxOthers = GetSubmittedOpCnt();
723 3 : HCCL_ERROR("SubmittedOpCnt: %u, OperatorParams: %s", idxOthers, opParams.Describe().c_str());
724 3 : HCCL_ERROR("Unknown error occurs!");
725 1 : return HcclResult::HCCL_E_INTERNAL;
726 1 : }
727 5 : return HcclResult::HCCL_SUCCESS;
728 : }
729 :
730 26 : HcclResult CommunicatorImpl::CheckCommStatus() const
731 : {
732 26 : if (GetCommStatus() == CommStatus::COMM_ERROR) {
733 12 : HCCL_ERROR("Comm has been error, can not load opbased operator now!");
734 4 : return HcclResult::HCCL_E_INTERNAL;
735 : }
736 :
737 22 : if (isSuspended) {
738 3 : HCCL_ERROR("Comm has been suspended, can not load opbased operator now!");
739 1 : return HcclResult::HCCL_E_SUSPENDING;
740 : }
741 21 : return HcclResult::HCCL_SUCCESS;
742 : }
743 :
744 6 : HcclResult CommunicatorImpl::AllocCollOpResource(const CollOpParams &opParams, void **addr)
745 : {
746 : try {
747 6 : if (opParams.commEngine != HcclAccelerator::AICPU_TS) {
748 3 : HCCL_ERROR("[CommunicatorImpl][%s] Only AICPU_TS is supported for aicpu unfold on mc2. input is %s", __func__, opParams.commEngine.Describe().c_str());
749 5 : return HCCL_E_NOT_SUPPORT;
750 : }
751 11 : CHK_RET(CheckCommStatus());
752 :
753 3 : WaitReady();
754 3 : curOpParams = opParams;
755 3 : CovertToCurrentCollOperator(id, opParams, OpMode::OPBASE, false);
756 3 : opExecuteConfig = commExecuteConfig;
757 3 : ExecAlgSelect(opParams, OpMode::OPBASE);
758 3 : CHK_PTR_NULL(collService);
759 3 : if (dynamic_cast<CollServiceDefaultImpl *>(collService) != nullptr) {
760 3 : HCCL_ERROR("The op base is not supported in expanding on the host in 950 with MC2.");
761 1 : return HcclResult::HCCL_E_NOT_SUPPORT;
762 : }
763 :
764 2 : SetCommStatus(CommStatus::COMM_READY);
765 5 : CHK_RET(OpParamsChecker::CheckOpDataTypeOpbase(opParams, GetOpCcuFeatureFlag(), GetOpAiCpuTSFeatureFlag(), false));
766 1 : SetCommStatus(CommStatus::COMM_INUSE);
767 1 : std::string opAlgTag = opParams.opTag + "_" + curAlgName;
768 1 : CHK_RET(collService->AllocCollOpResource(*currentCollOperator, opAlgTag, addr));
769 1 : SetCommStatus(CommStatus::COMM_READY);
770 1 : } catch (HcclException &e) {
771 0 : SetCommStatus(CommStatus::COMM_READY);
772 0 : HCCL_ERROR(e.what());
773 0 : PrintBackTrace(e);
774 0 : HCCL_ERROR("AllocCollOpResource OperatorParams: %s", opParams.Describe().c_str());
775 0 : return e.GetErrorCode();
776 0 : } catch (exception &e) {
777 0 : SetCommStatus(CommStatus::COMM_READY);
778 0 : HCCL_ERROR(e.what());
779 0 : HCCL_ERROR("AllocCollOpResource OperatorParams: %s", opParams.Describe().c_str());
780 0 : return HcclResult::HCCL_E_INTERNAL;
781 0 : } catch (...) {
782 0 : SetCommStatus(CommStatus::COMM_READY);
783 0 : HCCL_ERROR("AllocCollOpResource OperatorParams: %s", opParams.Describe().c_str());
784 0 : HCCL_ERROR("Unkown error occurs!");
785 0 : return HcclResult::HCCL_E_INTERNAL;
786 0 : }
787 1 : return HcclResult::HCCL_SUCCESS;
788 : }
789 :
790 1 : HcclResult CommunicatorImpl::CalcCollOffloadOpRes(const OpType opType, u64 dataSize, HcclDataType dataType, CollOffloadOpResReq &resReq)
791 : {
792 3 : HCCL_INFO("[CommunicatorImpl][%s] start, opType[%s], dataSize[%llu].", __func__, opType.Describe().c_str(),
793 : dataSize);
794 : try {
795 : // 资源计算
796 : HcclResult errCode
797 1 : = collAlgComponent->CalcResOffload(opType, dataSize, dataType, GetCommExecuteConfig(), resReq); // 通信域粒度
798 1 : if (errCode != HcclResult::HCCL_SUCCESS) {
799 : std::string msg
800 : = StringFormat("[CommunicatorImpl][%s] Error occurs when call collAlgComponent.CalcResOffload, "
801 : "error code: %d",
802 0 : __func__, errCode);
803 0 : HCCL_ERROR(msg.c_str());
804 0 : return errCode;
805 0 : }
806 0 : } catch (HcclException &e) {
807 0 : HCCL_ERROR(e.what());
808 0 : return e.GetErrorCode();
809 0 : } catch (exception &e) {
810 0 : HCCL_ERROR(e.what());
811 0 : return HcclResult::HCCL_E_INTERNAL;
812 0 : } catch (...) {
813 0 : HCCL_ERROR("Unknown error occurs!");
814 0 : return HcclResult::HCCL_E_INTERNAL;
815 0 : }
816 3 : HCCL_INFO("[CommunicatorImpl][%s] end.", __func__);
817 1 : return HcclResult::HCCL_SUCCESS;
818 : }
819 :
820 0 : HcclResult CommunicatorImpl::SetCollOffloadSlaveStreams(const std::string &opTag,
821 : std::vector<void *> slaveStreams)
822 : {
823 : try {
824 0 : HCCL_INFO("[CommunicatorImpl][%s] start, opTag[%s].", __func__, opTag.c_str());
825 : // 将slaveStreams注册到streamManager中
826 0 : RegisterOffloadSlaveStreams(opTag, slaveStreams);
827 0 : HCCL_INFO("[CommunicatorImpl][%s] end.", __func__);
828 0 : } catch (HcclException &e) {
829 0 : HCCL_ERROR(e.what());
830 0 : return e.GetErrorCode();
831 0 : } catch (exception &e) {
832 0 : HCCL_ERROR(e.what());
833 0 : return HcclResult::HCCL_E_INTERNAL;
834 0 : } catch (...) {
835 0 : HCCL_ERROR("Unknown error occurs!");
836 0 : return HcclResult::HCCL_E_INTERNAL;
837 0 : }
838 0 : return HcclResult::HCCL_SUCCESS;
839 : }
840 :
841 6 : HcclResult CommunicatorImpl::SetCollOffloadScratchBuf(const std::string &opTag,
842 : void *scratchMemPtr,
843 : u64 requiredScratchMemSize)
844 : {
845 : try {
846 18 : HCCL_INFO("[CommunicatorImpl][%s] start, opTag[%s] requiredScratchMemSize[%llu].", __func__, opTag.c_str(), requiredScratchMemSize);
847 : // 将scratchBuf注册到dataBufManager中
848 6 : RegisterOffloadScratchBuffer(opTag, scratchMemPtr, requiredScratchMemSize);
849 18 : HCCL_INFO("[CommunicatorImpl][%s] end.", __func__);
850 0 : } catch (HcclException &e) {
851 0 : HCCL_ERROR(e.what());
852 0 : return e.GetErrorCode();
853 0 : } catch (exception &e) {
854 0 : HCCL_ERROR(e.what());
855 0 : return HcclResult::HCCL_E_INTERNAL;
856 0 : } catch (...) {
857 0 : HCCL_ERROR("Unknown error occurs!");
858 0 : return HcclResult::HCCL_E_INTERNAL;
859 0 : }
860 6 : return HcclResult::HCCL_SUCCESS;
861 : }
862 :
863 0 : void CommunicatorImpl::RegisterOffloadSlaveStreams(const std::string &opTag, std::vector<void *> slaveStreams) const
864 : {
865 0 : StreamManager &sm = GetStreamManager();
866 0 : sm.offload->RegisterSlaves(opTag, slaveStreams);
867 0 : }
868 :
869 6 : void CommunicatorImpl::RegisterOffloadScratchBuffer(const std::string &opTag, void *scratchMemPtr,
870 : u64 requiredScratchMemSize)
871 : {
872 6 : auto scratchBuffer = DevBuffer::Create(reinterpret_cast<uintptr_t>(scratchMemPtr), requiredScratchMemSize);
873 6 : if(scratchBuffer){
874 6 : offloadScrachBufferMap[opTag] = scratchBuffer;
875 18 : HCCL_RUN_INFO("[CommunicatorImpl] offloadScratchBuffer register, opTag[%s], offloadScrachBufferAddr[%llu], "
876 : "offloadScrachBufferBufSize[%llu]M",
877 : opTag.c_str(), scratchBuffer->GetAddr(),
878 : scratchBuffer->GetSize() / HCCL_CCL_COMM_FIXED_CALC_BUFFER_SIZE);
879 : }
880 6 : }
881 :
882 8 : HcclResult CommunicatorImpl::LoadOffloadCollOp(std::string &opTag, const CollOpParams &opParams, void *stream)
883 : {
884 : try {
885 24 : HCCL_INFO("CommunicatorImpl::LoadOffloadCollOp dataType[%s]", opParams.dataType.Describe().c_str());
886 8 : isLoadOp = true;
887 8 : curOpParams = opParams;
888 8 : if (GetCommStatus() == CommStatus::COMM_ERROR) {
889 9 : HCCL_ERROR("Comm has been error, can not offload operator now!");
890 7 : return HcclResult::HCCL_E_INTERNAL;
891 : }
892 :
893 5 : if (isSuspended) {
894 0 : HCCL_ERROR("Comm has been suspended, can not offload operator now!");
895 0 : return HcclResult::HCCL_E_SUSPENDING;
896 : }
897 :
898 : // 等待通信域状态为Ready，执行算子下发
899 5 : WaitReady();
900 5 : SnapShotParser::GetInstance().SetIsNeedLoadOp(false);
901 5 : if (rankSize == 1) {
902 3 : HCCL_WARNING("[CommunicatorImpl][%s] ranksize == 1, enter SingleRankProc", __func__);
903 1 : SingleRankProc(opParams, stream);
904 1 : return HcclResult::HCCL_SUCCESS;
905 : }
906 4 : uint64_t beginTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
907 :
908 : // 更新开关状态
909 4 : UpdateProfStat();
910 : // 判断是否为aclgraph(aicpu场景零拷贝会切图模式)
911 4 : bool isCapture = false; // isCapture为true表示aclgraph，profiling需要
912 4 : CHK_RET(MatchAclgraph(stream, isCapture));
913 12 : HCCL_INFO("CommunicatorImpl::LoadOffloadCollOp opParams dataType[%s]", opParams.dataType.Describe().c_str());
914 4 : CovertToCurrentCollOperator(opTag, opParams, OpMode::OFFLOAD);
915 12 : HCCL_INFO("CommunicatorImpl::LoadOffloadCollOp currentCollOperator dataType[%s]", currentCollOperator->dataType.Describe().c_str());
916 : // 图模式算子加载选择CollService
917 4 : opExecuteConfig = commExecuteConfig;
918 4 : ExecAlgSelect(opParams, OpMode::OFFLOAD);
919 :
920 4 : if (opExecuteConfig.accState == AcceleratorState::HOSTCPU_TS) { // 950不支持HOST_TS模式
921 3 : HCCL_ERROR("[CommunicatorImpl::LoadOffloadCollOp] HOSTCPU_TS is not support.");
922 1 : return HcclResult::HCCL_E_NOT_SUPPORT;
923 : }
924 :
925 3 : bool isAiv = (opExecuteConfig.accState == AcceleratorState::AIV || opExecuteConfig.accState == AcceleratorState::AIV_ONLY);
926 9 : CHK_RET(OpParamsChecker::CheckOpDataTypeOffload(opParams, GetOpCcuFeatureFlag(), GetOpAiCpuTSFeatureFlag(), isAiv)); // 算子粒度
927 :
928 1 : if (isAiv) {
929 0 : currentCollOperator->numBlocksLimit = aivCoreLimit;
930 0 : HCCL_INFO("[CommunicatorImpl::LoadOffloadCollOp] Aiv core limit is [%u].", aivCoreLimit);
931 : }
932 1 : if (isAiv && aivClearEnable) {
933 0 : aivOffloadTag = 1;
934 1 : } else if (isAiv) {
935 0 : aivOffloadTag++;
936 : }
937 :
938 : // 避免transport建链前，通讯域被摧毁
939 1 : SetCommStatus(CommStatus::COMM_INUSE);
940 1 : collService->LoadWithOffloadMode(*currentCollOperator, std::make_unique<Stream>(stream));
941 1 : SetCommStatus(CommStatus::COMM_READY);
942 1 : bool cachedReq = opParams.staticShape || isCapture;
943 1 : ReportProfInfo(beginTime, cachedReq, isCapture); // profiling对于aclgraph场景的处理与单算子一致
944 1 : opIndex++;
945 0 : } catch (HcclException &e) {
946 0 : SetCommStatus(CommStatus::COMM_READY);
947 0 : HCCL_ERROR(e.what());
948 0 : return e.GetErrorCode();
949 0 : } catch (exception &e) {
950 0 : SetCommStatus(CommStatus::COMM_READY);
951 0 : HCCL_ERROR(e.what());
952 0 : return HcclResult::HCCL_E_INTERNAL;
953 0 : } catch (...) {
954 0 : SetCommStatus(CommStatus::COMM_READY);
955 0 : HCCL_ERROR("Unknown error occurs!");
956 0 : return HcclResult::HCCL_E_INTERNAL;
957 0 : }
958 1 : return HcclResult::HCCL_SUCCESS;
959 : }
960 :
961 19 : void CommunicatorImpl::CalcA2ASendRecvMem(const CollOpParams &opParams, u64 &sendSize, u64 &recvSize, bool isHcomSelectAlg) const
962 : {
963 19 : u64 sendCount = 0;
964 19 : u64 recvCount = 0;
965 19 : u32 sendTypeSize = 0;
966 19 : u32 recvTypeSize = 0;
967 19 : if (opParams.opType == OpType::ALLTOALLV && !isHcomSelectAlg) { // alltoallv场景hcomSeletAlg不填写alltoallv的参数
968 2 : for (u32 i = 0; i < rankSize; i++) {
969 1 : CHECK_NULLPTR((static_cast<const u64 *>(opParams.all2AllVDataDes.sendCounts) + i),
970 2 : StringFormat("%s failed, opParams.all2AllVDataDes.sendCounts[%u] is nullptr", __func__, i));
971 1 : CHECK_NULLPTR((static_cast<const u64 *>(opParams.all2AllVDataDes.sdispls) + i),
972 2 : StringFormat("%s failed, opParams.all2AllVDataDes.sdispls[%u] is nullptr", __func__, i));
973 1 : CHECK_NULLPTR((static_cast<const u64 *>(opParams.all2AllVDataDes.recvCounts) + i),
974 2 : StringFormat("%s failed, opParams.all2AllVDataDes.recvCounts[%u] is nullptr", __func__, i));
975 1 : CHECK_NULLPTR((static_cast<const u64 *>(opParams.all2AllVDataDes.rdispls) + i),
976 2 : StringFormat("%s failed, opParams.all2AllVDataDes.rdispls[%u] is nullptr", __func__, i));
977 1 : u64 curSendCount = *(static_cast<const u64 *>(opParams.all2AllVDataDes.sendCounts) + i) +
978 1 : *(static_cast<const u64 *>(opParams.all2AllVDataDes.sdispls) + i);
979 1 : sendCount = std::max(sendCount, curSendCount);
980 1 : u64 curRecvCount = *(static_cast<const u64 *>(opParams.all2AllVDataDes.recvCounts) + i) +
981 1 : *(static_cast<const u64 *>(opParams.all2AllVDataDes.rdispls) + i);
982 1 : recvCount = std::max(recvCount, curRecvCount);
983 : }
984 1 : sendTypeSize = DataTypeSizeGet(opParams.all2AllVDataDes.sendType);
985 1 : recvTypeSize = DataTypeSizeGet(opParams.all2AllVDataDes.recvType);
986 18 : } else if (opParams.opType == OpType::ALLTOALLVC && !isHcomSelectAlg) { // alltoallvc场景hcomSeletAlg不填写alltoallvc的参数
987 2 : for (u32 i = 0; i < rankSize; i++) {
988 1 : CHECK_NULLPTR((static_cast<const u64 *>(opParams.all2AllVCDataDes.sendCountMatrix) + myRank * rankSize + i),
989 2 : StringFormat("%s failed, opParams.all2AllVCDataDes.sendCountMatrix[%u] is nullptr", __func__, (myRank * rankSize + i)));
990 1 : sendCount += *(static_cast<const u64 *>(opParams.all2AllVCDataDes.sendCountMatrix) +
991 1 : myRank * rankSize + i);
992 1 : recvCount += *(static_cast<const u64 *>(opParams.all2AllVCDataDes.sendCountMatrix) +
993 1 : myRank + rankSize * i);
994 : }
995 1 : sendTypeSize = DataTypeSizeGet(opParams.all2AllVCDataDes.sendType);
996 1 : recvTypeSize = DataTypeSizeGet(opParams.all2AllVCDataDes.recvType);
997 : } else {
998 17 : sendCount = opParams.all2AllDataDes.sendCount * rankSize;
999 17 : recvCount = opParams.all2AllDataDes.recvCount * rankSize;
1000 17 : sendTypeSize = DataTypeSizeGet(opParams.all2AllDataDes.sendType);
1001 17 : recvTypeSize = DataTypeSizeGet(opParams.all2AllDataDes.recvType);
1002 : }
1003 19 : sendSize = sendCount * sendTypeSize;
1004 19 : recvSize = recvCount * recvTypeSize;
1005 19 : }
1006 :
1007 19 : void CommunicatorImpl::ConvertCollOperatorA2A(const CollOpParams &opParams, bool isLaunch, bool isHcomSelectAlg)
1008 : {
1009 19 : if (currentCollOperator == nullptr) {
1010 0 : std::string msg = StringFormat("currentCollOperator is nullptr");
1011 0 : THROW<NullPtrException>(msg);
1012 0 : }
1013 :
1014 19 : if (isLaunch) {
1015 19 : LaunchConvertCollOperatorA2A(opParams, isHcomSelectAlg);
1016 : } else {
1017 0 : DefaultConvertCollOperatorA2A(opParams);
1018 : }
1019 19 : }
1020 :
1021 0 : void CommunicatorImpl::DefaultConvertCollOperatorA2A(const CollOpParams &opParams)
1022 : {
1023 : // MC2场景、图模式算法选择场景准备资源场景下只需默认值
1024 0 : HCCL_INFO("DefaultConvertCollOperatorA2A start.");
1025 0 : if (opParams.opType == OpType::ALLTOALL) {
1026 0 : currentCollOperator->all2AllDataDes.sendCount = 0;
1027 0 : currentCollOperator->all2AllDataDes.recvCount = 0;
1028 0 : currentCollOperator->all2AllDataDes.sendType = DataType::FP16;
1029 0 : currentCollOperator->all2AllDataDes.recvType = DataType::FP16;
1030 0 : currentCollOperator->dataType = DataType::FP16;
1031 0 : } else if (opParams.opType == OpType::ALLTOALLV) {
1032 0 : currentCollOperator->all2AllVDataDes.sendType = DataType::FP16;
1033 0 : currentCollOperator->all2AllVDataDes.recvType = DataType::FP16;
1034 0 : currentCollOperator->dataType = DataType::FP16;
1035 0 : } else if (opParams.opType == OpType::ALLTOALLVC) {
1036 0 : currentCollOperator->all2AllVCDataDes.sendType = DataType::FP16;
1037 0 : currentCollOperator->all2AllVCDataDes.recvType = DataType::FP16;
1038 0 : currentCollOperator->dataType = DataType::FP16;
1039 : }
1040 0 : }
1041 :
1042 19 : void CommunicatorImpl::LaunchConvertCollOperatorA2A(const CollOpParams &opParams, bool isHcomSelectAlg)
1043 : {
1044 : // 下发算子场景下需要继承值并准备Mem
1045 57 : HCCL_INFO("LaunchConvertCollOperatorA2A start.");
1046 19 : if (opParams.opType == OpType::ALLTOALL) {
1047 9 : currentCollOperator->all2AllDataDes.sendCount = opParams.all2AllDataDes.sendCount;
1048 9 : currentCollOperator->all2AllDataDes.recvCount = opParams.all2AllDataDes.recvCount;
1049 9 : currentCollOperator->all2AllDataDes.sendType = opParams.all2AllDataDes.sendType;
1050 9 : currentCollOperator->all2AllDataDes.recvType = opParams.all2AllDataDes.recvType;
1051 9 : currentCollOperator->dataType = opParams.all2AllDataDes.sendType;
1052 27 : HCCL_INFO("sendCount[%llu], recvCount[%llu]", opParams.all2AllDataDes.sendCount, opParams.all2AllDataDes.recvCount);
1053 10 : } else if (opParams.opType == OpType::ALLTOALLV) {
1054 7 : currentCollOperator->all2AllVDataDes.sendCounts = opParams.all2AllVDataDes.sendCounts;
1055 7 : currentCollOperator->all2AllVDataDes.recvCounts = opParams.all2AllVDataDes.recvCounts;
1056 7 : currentCollOperator->all2AllVDataDes.sdispls = opParams.all2AllVDataDes.sdispls;
1057 7 : currentCollOperator->all2AllVDataDes.rdispls = opParams.all2AllVDataDes.rdispls;
1058 7 : currentCollOperator->all2AllVDataDes.sendType = opParams.all2AllVDataDes.sendType;
1059 7 : currentCollOperator->all2AllVDataDes.recvType = opParams.all2AllVDataDes.recvType;
1060 7 : currentCollOperator->dataType = opParams.all2AllVDataDes.sendType;
1061 3 : } else if (opParams.opType == OpType::ALLTOALLVC) {
1062 3 : currentCollOperator->all2AllVCDataDes.sendType = opParams.all2AllVCDataDes.sendType;
1063 3 : currentCollOperator->all2AllVCDataDes.recvType = opParams.all2AllVCDataDes.recvType;
1064 3 : currentCollOperator->all2AllVCDataDes.sendCountMatrix = opParams.all2AllVCDataDes.sendCountMatrix;
1065 3 : currentCollOperator->dataType = opParams.all2AllVCDataDes.sendType;
1066 : }
1067 :
1068 19 : u64 sendSize = 0;
1069 19 : u64 recvSize = 0;
1070 19 : CalcA2ASendRecvMem(opParams, sendSize, recvSize, isHcomSelectAlg);
1071 57 : HCCL_INFO("sendSize[%llu], recvSize[%llu]", sendSize, recvSize);
1072 19 : currentCollOperator->inputMem = DevBuffer::Create(reinterpret_cast<uintptr_t >(opParams.sendBuf), sendSize);
1073 19 : currentCollOperator->outputMem = DevBuffer::Create(reinterpret_cast<uintptr_t >(opParams.recvBuf), recvSize);
1074 19 : }
1075 :
1076 51 : void CommunicatorImpl::ConvertCollOperatorMem(const CollOpParams &opParams, u64 size)
1077 : {
1078 153 : HCCL_INFO("[CommunicatorImpl][%s] start, opType[%s], size[%llu]", __func__, opParams.opType.Describe().c_str(), size);
1079 :
1080 51 : if (opParams.opType == OpType::REDUCESCATTER || opParams.opType == OpType::SCATTER) {
1081 10 : currentCollOperator->inputMem = DevBuffer::Create(reinterpret_cast<uintptr_t>(opParams.sendBuf), size * rankSize);
1082 : } else {
1083 41 : currentCollOperator->inputMem = DevBuffer::Create(reinterpret_cast<uintptr_t>(opParams.sendBuf), size);
1084 : }
1085 :
1086 51 : if (opParams.opType == OpType::ALLGATHER || opParams.opType == OpType::GATHER) {
1087 7 : currentCollOperator->outputMem = DevBuffer::Create(reinterpret_cast<uintptr_t>(opParams.recvBuf), size * rankSize);
1088 : } else {
1089 44 : currentCollOperator->outputMem = DevBuffer::Create(reinterpret_cast<uintptr_t>(opParams.recvBuf), size);
1090 : }
1091 :
1092 153 : HCCL_INFO("[CommunicatorImpl][%s] end.", __func__);
1093 51 : }
1094 :
1095 6 : void CommunicatorImpl::ConvertCollOperatorMemV(const CollOpParams &opParams, bool isHcomSelectAlg)
1096 : {
1097 6 : if (isHcomSelectAlg) {
1098 4 : return; // isHcomSeletAlg表示是否为图插件接口进来，若是跳过该步。未来aiv支持reducescatterv/allgatherv算子时，改处需做对应适配。
1099 : }
1100 6 : HCCL_INFO("[CommunicatorImpl::%s] start, opType[%s]", __func__, opParams.opType.Describe().c_str());
1101 2 : u64 size = DataTypeSizeGet(opParams.dataType) * opParams.count;
1102 :
1103 2 : u64 *counts = static_cast<u64 *>(opParams.vDataDes.counts);
1104 2 : u64 totalCount = 0;
1105 6 : for (size_t index = 0; index < rankSize; index++) {
1106 4 : totalCount += counts[index];
1107 : }
1108 2 : u64 totalSize = DataTypeSizeGet(opParams.dataType) * totalCount;
1109 :
1110 2 : if (opParams.opType == OpType::REDUCESCATTERV) {
1111 1 : currentCollOperator->inputMem = DevBuffer::Create(reinterpret_cast<uintptr_t>(opParams.sendBuf), totalSize);
1112 : } else {
1113 1 : currentCollOperator->inputMem = DevBuffer::Create(reinterpret_cast<uintptr_t>(opParams.sendBuf), size);
1114 : }
1115 :
1116 2 : if (opParams.opType == OpType::ALLGATHERV) {
1117 1 : currentCollOperator->outputMem = DevBuffer::Create(reinterpret_cast<uintptr_t>(opParams.recvBuf), totalSize);
1118 : } else {
1119 1 : currentCollOperator->outputMem = DevBuffer::Create(reinterpret_cast<uintptr_t>(opParams.recvBuf), size);
1120 : }
1121 :
1122 6 : HCCL_INFO("[CommunicatorImpl::%s] end.", __func__);
1123 : }
1124 :
1125 86 : void CommunicatorImpl::CovertToCurrentCollOperator(std::string &opTag, const CollOpParams &opParams, OpMode opMode, bool isLaunch, bool isHcomSelectAlg)
1126 : {
1127 86 : std::string errorMsg = "CovertToCurrentCollOperator make_unique<CollOperator> failed";
1128 86 : TRY_CATCH_THROW(InternalException, errorMsg, currentCollOperator = make_unique<CollOperator>(););
1129 86 : CHECK_NULLPTR(currentCollOperator, StringFormat("[CommunicatorImpl][%s] currentCollOperator is nullptr", __func__));
1130 86 : currentCollOperator->opMode = opMode;
1131 86 : currentCollOperator->opTag = opTag; // 单算子 标签 为通信域id, 图模式 标签 为传入的opTag
1132 86 : currentCollOperator->staticAddr = opParams.staticAddr;
1133 86 : currentCollOperator->staticShape = opParams.staticShape;
1134 86 : currentCollOperator->myRank = GetMyRank();
1135 86 : if (opMode == OpMode::OPBASE) { // 单算子Scratch buffer为CCL Buffer
1136 19 : currentCollOperator->scratchMem = DevBuffer::Create(GetCclBuffer()->GetAddr(), GetCclBuffer()->GetSize());
1137 67 : } else if (opMode == OpMode::OFFLOAD) {
1138 67 : if (offloadScrachBufferMap.find(opTag) != offloadScrachBufferMap.end()) {
1139 0 : auto scratchMem = offloadScrachBufferMap[opTag];
1140 0 : HCCL_INFO("[CommunicatorImpl::CovertToCurrentCollOperator] offloadScrachBufferMap[%s] is [%s]",
1141 : opTag.c_str(), scratchMem->Describe().c_str());
1142 0 : currentCollOperator->scratchMem = scratchMem;
1143 0 : }
1144 : }
1145 :
1146 86 : currentCollOperator->opType = opParams.opType;
1147 86 : currentCollOperator->reduceOp = opParams.reduceOp;
1148 86 : currentCollOperator->root = opParams.root;
1149 86 : currentCollOperator->outputDataType = opParams.outputDataType;
1150 86 : currentCollOperator->sendRecvRemoteRank = opParams.dstRank;
1151 86 : if (opParams.opType == OpType::ALLTOALL || opParams.opType == OpType::ALLTOALLV || opParams.opType == OpType::ALLTOALLVC) {
1152 17 : ConvertCollOperatorA2A(opParams, isLaunch, isHcomSelectAlg);
1153 69 : } else if (opParams.opType == OpType::BATCHSENDRECV) {
1154 0 : currentCollOperator->batchSendRecvDataDes.sendRecvItemsPtr = opParams.batchSendRecvDataDes.sendRecvItemsPtr;
1155 0 : currentCollOperator->batchSendRecvDataDes.itemNum = opParams.batchSendRecvDataDes.itemNum;
1156 0 : currentCollOperator->dataType = HcclDataTypeToDataType(static_cast<HcclSendRecvItem*>(opParams.batchSendRecvDataDes.sendRecvItemsPtr)->dataType);
1157 : } else {
1158 69 : currentCollOperator->dataType = opParams.dataType;
1159 69 : currentCollOperator->dataCount = opParams.count;
1160 69 : if(opParams.opType == OpType::REDUCESCATTERV || opParams.opType == OpType::ALLGATHERV){
1161 6 : currentCollOperator->vDataDes.counts = opParams.vDataDes.counts;
1162 6 : currentCollOperator->vDataDes.displs = opParams.vDataDes.displs;
1163 6 : currentCollOperator->vDataDes.dataType = opParams.vDataDes.dataType;
1164 6 : ConvertCollOperatorMemV(opParams, isHcomSelectAlg);
1165 : } else {
1166 63 : u64 size = DataTypeSizeGet(opParams.dataType) * opParams.count;
1167 63 : if (size != 0) {
1168 51 : ConvertCollOperatorMem(opParams, size);
1169 : } else {
1170 36 : HCCL_WARNING("[CommunicatorImpl::%s] size is 0", __func__);
1171 : }
1172 : }
1173 : }
1174 258 : HCCL_INFO("CommunicatorImpl::%s op dataType[%s], dataCount[%llu]", __func__, currentCollOperator->dataType.Describe().c_str(), currentCollOperator->dataCount);
1175 86 : }
1176 :
1177 47 : void CommunicatorImpl::InitCommonData(const CommParams &commParams, const HcclCommConfig &commConfig)
1178 : {
1179 47 : InitCommonDataNotInitDevType(commParams, commConfig);
1180 : // 设定devType，初始化能力，算法及其他模块通过Get获取能力
1181 47 : DevCapability::GetInstance().Init(devType);
1182 47 : }
1183 :
1184 49 : void CommunicatorImpl::InitCommonDataNotInitDevType(const CommParams &commParams, const HcclCommConfig &commConfig)
1185 : {
1186 49 : InitCommonData(commParams);
1187 49 : config = commConfig;
1188 49 : cclBufferSize = config.hcclBufferSize;
1189 49 : }
1190 :
1191 50 : void CommunicatorImpl::InitCommonData(const CommParams &commParams)
1192 : {
1193 50 : id = commParams.commId;
1194 50 : idIndex = globalIndex.fetch_add(1);
1195 50 : establishLinkSocketTag = id + "_establish_link" + "_" + "exchanger";
1196 50 : myRank = commParams.myRank;
1197 50 : rankSize = commParams.rankSize;
1198 50 : rankInParentComm = commParams.rankInParentComm;
1199 50 : devType = commParams.devType;
1200 50 : isWorldGroup = commParams.isWorldGroup;
1201 50 : devLogicId = HrtGetDevice();
1202 50 : devPhyId = HrtGetDevicePhyIdByIndex(devLogicId);
1203 50 : }
1204 :
1205 7 : void CommunicatorImpl::CheckRankGraph() const
1206 : {
1207 : // 校验虚拟拓扑中的rankSize和通信域的rankSize一致
1208 7 : u32 virtRankSize = rankGraph->GetRankSize();
1209 7 : if (virtRankSize != rankSize) {
1210 : std::string msg
1211 : = StringFormat("Check rankGraph failed, communicator rankSize[%u] does not equal rankTable rankSize[%u]",
1212 1 : rankSize, virtRankSize);
1213 1 : THROW<InvalidParamsException>(msg);
1214 1 : }
1215 :
1216 : // 校验0值
1217 6 : u32 num = rankGraph->GetInnerRankSize();
1218 6 : if (num == 0) {
1219 : std::string msg
1220 : = StringFormat("Check rankGraph failed, inner rankSize should not be %u",
1221 1 : num);
1222 1 : THROW<InvalidParamsException>(msg);
1223 1 : }
1224 5 : CheckRankGraphAddrs();
1225 5 : }
1226 :
1227 5 : void CommunicatorImpl::CheckRankGraphAddrs() const
1228 : {
1229 5 : if (rankGraph == nullptr || ranktableInfo == nullptr) {
1230 0 : HCCL_WARNING("[CommunicatorImpl][%s] rankGraph or ranktableInfo is nullptr, skip.", __func__);
1231 5 : return;
1232 : }
1233 :
1234 5 : if (rankGraph->GetRankSize() == 1) {
1235 0 : HCCL_WARNING("[CommunicatorImpl][%s] single rank no need do this check!", __func__);
1236 0 : return;
1237 : }
1238 :
1239 : // 仅能获取到当前进程所在卡的ip，每个卡独立check自己的部分
1240 5 : std::unordered_set<Eid> localEidSet;
1241 5 : NewRankInfo localRankInfo;
1242 10 : for (auto &rank : ranktableInfo->ranks) {
1243 10 : if (rank.deviceId == devPhyId) { // 获取本卡的ip地址
1244 5 : HRaInfo info(HrtNetworkMode::HDC, rank.deviceId);
1245 5 : std::vector<HrtDevEidInfo> localEidInfos = HrtRaGetDevEidInfoList(info);
1246 5 : for (auto &eidInfo : localEidInfos) {
1247 0 : localEidSet.insert(eidInfo.ipAddress.GetEid());
1248 : }
1249 5 : localRankInfo = rank;
1250 5 : break;
1251 5 : }
1252 : }
1253 :
1254 5 : if (localEidSet.empty()) {
1255 5 : return;
1256 : }
1257 :
1258 0 : const std::shared_ptr<NetInstance::Peer> &peer = rankGraph->GetPeer(myRank);
1259 0 : const std::vector<std::shared_ptr<NetInstance::ConnInterface>> &interfaces = peer->GetIfaces();
1260 0 : for(auto &interface : interfaces) {
1261 0 : const std::set<LinkProtocol> &protocols = interface->GetLinkProtocols(); // PCIE和UBOE没有EID
1262 0 : if (interface->GetPos() == AddrPosition::DEVICE && protocols.count(LinkProtocol::PCIE) == 0
1263 0 : && protocols.count(LinkProtocol::UBOE) == 0 && localEidSet.count(interface->GetAddr().GetEid()) == 0) {
1264 0 : RPT_INPUT_ERR(true, "EI0014", std::vector<std::string>({"value", "variable", "expect"}),
1265 : std::vector<std::string>({interface->GetAddr().GetIpStr(), "addr", "A right ip address"}));
1266 0 : THROW<InvalidParamsException>(StringFormat("[CommunicatorImpl][%s]"
1267 : "the ip address %s of ranktable in rank %u is error!",
1268 0 : __func__, interface->GetAddr().Describe().c_str(), devPhyId));
1269 : }
1270 0 : }
1271 10 : }
1272 :
1273 :
1274 40 : u32 GetLocalDieId(PortData&& port, LinkProtocol linkProtocol)
1275 : {
1276 40 : auto devLogicId = HrtGetDevice();
1277 40 : uint32_t devPhyId = HrtGetDevicePhyIdByIndex(devLogicId);
1278 :
1279 40 : auto &rdmaHandleMgr = RdmaHandleManager::GetInstance();
1280 40 : auto rdmaHandle = rdmaHandleMgr.Get(devPhyId, port, linkProtocol);
1281 40 : auto dieId = rdmaHandleMgr.GetDieAndFuncId(rdmaHandle).first;
1282 40 : return dieId;
1283 : }
1284 :
1285 : constexpr u32 localPortId = 0;
1286 :
1287 12 : void CommunicatorImpl::InitRankGraph(const string &ranktableM)
1288 : {
1289 : JsonParser rankTableParser{};
1290 12 : RankTableInfo rankTableInfo{};
1291 12 : rankTableParser.ParseString(ranktableM, rankTableInfo);
1292 5 : InitRankGraph(rankTableInfo);
1293 12 : }
1294 :
1295 0 : std::string CommunicatorImpl::GetTopoFilePath()
1296 : {
1297 0 : HCCL_INFO("[CommunicatorImpl::%s] start.", __func__);
1298 :
1299 0 : std::string filePath = "/etc/hccl_rootinfo.json";
1300 : JsonParser jsonParser{};
1301 0 : nlohmann::json parseJson{};
1302 0 : std::string topoFilePath{};
1303 0 : std::ifstream file(filePath);
1304 0 : if (file.good()) {
1305 0 : jsonParser.ParseFileToJson(filePath, parseJson);
1306 0 : std::string msgRankTopoFile = "error occurs when parser object of propName \"topo_file_path\"";
1307 0 : TRY_CATCH_THROW(InvalidParamsException, msgRankTopoFile, topoFilePath = GetJsonProperty(parseJson, "topo_file_path"););
1308 0 : } else {
1309 0 : const size_t bufSize = 1024;
1310 0 : auto devLogicId = HrtGetDevice();
1311 0 : auto devPhyId = HrtGetDevicePhyIdByIndex(devLogicId);
1312 0 : std::vector<char> buffer(bufSize, '\0');
1313 0 : int result = TopoAddrInfoGetTopoFilePath(devPhyId, buffer.data(), buffer.size());
1314 0 : CHK_PRT_THROW(result != 0,
1315 : HCCL_ERROR("[%s] Get topo file path failed.", __func__),
1316 : InvalidParamsException, "Get topo file path failed.");
1317 0 : topoFilePath = std::string(buffer.data());
1318 0 : }
1319 :
1320 : // check topo_file_path
1321 0 : char resolvedPath[PATH_MAX] = {0};
1322 0 : CHK_PRT_THROW(realpath(topoFilePath.c_str(), resolvedPath) == nullptr,
1323 : HCCL_ERROR("[%s] topo_file_path[%s] is not a valid real path", __func__, topoFilePath.c_str()),
1324 : InvalidParamsException, "topo_file_path error");
1325 0 : return topoFilePath;
1326 0 : }
1327 :
1328 5 : void CommunicatorImpl::InitRankGraph(const RankTableInfo &ranktable)
1329 : {
1330 5 : string topoPath = GetTopoFilePath();
1331 5 : RankGraphBuilder rankGraphBuilder;
1332 5 : rankGraph = rankGraphBuilder.Build(ranktable, topoPath, myRank);
1333 5 : ranktableInfo = rankGraphBuilder.GetRankTableInfo(); // 获取ranktable信息
1334 15 : HCCL_RUN_INFO("[CommunicatorImpl::%s] rankTableInfo: %s", __func__, ranktableInfo->Describe().c_str());
1335 5 : topoInfo = rankGraphBuilder.GetTopoInfo(); // 获取topo信息
1336 15 : HCCL_RUN_INFO("[CommunicatorImpl][InitRankGraph] topoInfo[%s]", topoInfo->Describe().c_str());
1337 5 : rankSize = rankGraph->GetRankSize();
1338 5 : CheckRankGraph();
1339 5 : SaveTopoDesc(id);
1340 5 : std::vector<LinkData> fullLinks = GetFullMeshLinks();
1341 20 : for (auto link : fullLinks) {
1342 45 : HCCL_RUN_INFO("[CommunicatorImpl][InitRankGraph] link[%s]", link.Describe().c_str());
1343 : }
1344 5 : }
1345 :
1346 2 : void CommunicatorImpl::InitRankGraph(std::unique_ptr<RankGraph> &inputRankGraph)
1347 : {
1348 2 : if (inputRankGraph != nullptr) {
1349 2 : rankGraph = std::move(inputRankGraph);
1350 : } else {
1351 0 : std::string msg = StringFormat("Init RankGraph failed, inputRankGraph is nullptr");
1352 0 : THROW<NullPtrException>(msg);
1353 0 : }
1354 2 : CheckRankGraph();
1355 0 : SaveTopoDesc(id);
1356 0 : }
1357 :
1358 40 : void CommunicatorImpl::InitDataBufferManager()
1359 : {
1360 : // 申请scratchMem
1361 40 : u64 scratchBufSize = static_cast<u64>(GetBufferSize());
1362 40 : if (scratchBufSize == 0) {
1363 17 : scratchBufSize = EnvConfig::GetInstance().GetAlgoConfig().GetBuffSize();
1364 : } else {
1365 23 : scratchBufSize = scratchBufSize * HCCL_CCL_COMM_FIXED_CALC_BUFFER_SIZE;
1366 : }
1367 : // 如果是自定义算子流程，cclBufferSize的大小为2倍
1368 40 : scratchBufSize = scratchBufSize * INDEPENDENT_OP_BUFFER_SIZE_TIMES;
1369 40 : cclBufferSize = scratchBufSize;
1370 :
1371 : // aiv mc2预埋1M，并不暴露在内部算子执行逻辑里
1372 40 : scratchBufSize += HCCL_MC2_ON_AICPU_FIXED_CALC_BUFFER_SIZE;
1373 :
1374 40 : if (rankSize > 1) {
1375 39 : aivOffloadTagBuffer = std::move(DevBuffer::CreateHugePageBuf(HCCL_AIV_OFFLOAD_TAG_BUFFER_SIZE));
1376 39 : HrtMemset(reinterpret_cast<void*>(aivOffloadTagBuffer->GetAddr()), aivOffloadTagBuffer->GetSize(), aivOffloadTagBuffer->GetSize());
1377 39 : cclBuffer = std::move(DevBuffer::CreateHugePageBuf(scratchBufSize));
1378 117 : HCCL_RUN_INFO(
1379 : "[CommunicatorImpl][InitDataBufferManager] cclBuffer create, commId[%s], addr[%llu], size[%llu]M",
1380 : GetId().c_str(), cclBuffer->GetAddr(), cclBufferSize / HCCL_CCL_COMM_FIXED_CALC_BUFFER_SIZE);
1381 :
1382 39 : u64 aivTagBufSize = HCCL_CCL_AIV_TAG_BUFFER_SIZE * HCCL_CCL_COMM_FIXED_CALC_BUFFER_SIZE;
1383 117 : HCCL_INFO("[CommunicatorImpl][InitDataBufferManager] aivTagBufSize[%llu]M", aivTagBufSize / HCCL_CCL_COMM_FIXED_CALC_BUFFER_SIZE);
1384 39 : aivTagBuffer = std::move(DevBuffer::CreateHugePageBuf(aivTagBufSize));
1385 39 : HrtMemset(reinterpret_cast<void*>(aivTagBuffer->GetAddr()), aivTagBuffer->GetSize(), aivTagBuffer->GetSize());
1386 : }
1387 40 : dataBufferManager = std::make_unique<DataBufManager>();
1388 :
1389 40 : localRmaBufManager = std::make_unique<LocalRmaBufManager>(*this);
1390 :
1391 40 : remoteRmaBufManager = std::make_unique<RemoteRmaBufManager>(*this);
1392 40 : }
1393 :
1394 242 : void CommunicatorImpl::InitNotifyManager()
1395 : {
1396 242 : aicpuQueueNotifyManager_ = std::make_unique<QueueNotifyManager>(*this);
1397 :
1398 242 : ccuQueueNotifyManager_ = std::make_unique<QueueNotifyManager>(*this);
1399 :
1400 242 : queueWaitGroupCntNotifyManager = std::make_unique<QueueWaitGroupCntNotifyManager>();
1401 :
1402 242 : queueBcastPostCntNotifyManager = std::make_unique<QueueBcastPostCntNotifyManager>();
1403 :
1404 242 : connLocalNotifyManager = std::make_unique<ConnLocalNotifyManager>(this);
1405 :
1406 242 : connLocalCntNotifyManager = std::make_unique<ConnLocalCntNotifyManager>(this);
1407 :
1408 242 : ccuStreamSyncNotifyManager = std::make_unique<CcuStreamSyncNotifyManager>();
1409 242 : }
1410 :
1411 248 : void CommunicatorImpl::InitStreamManager()
1412 : {
1413 248 : streamManager = std::make_unique<StreamManager>(this);
1414 248 : aicpuStreamManager = std::make_unique<AicpuStreamManager>();
1415 248 : }
1416 :
1417 193 : void CommunicatorImpl::InitCollService()
1418 : {
1419 579 : HCCL_INFO("CommunicatorImpl::InitCollServices start");
1420 :
1421 193 : auto ccuCollService = std::make_shared<CollServiceDeviceMode>(this); // host 展开，ccu使用
1422 193 : auto aiCpuCollService = std::make_shared<CollServiceAiCpuImpl>(this); // aicpu 展开
1423 193 : auto hostCollService = std::make_shared<CollServiceDefaultImpl>(this); // host 展开，图模式使用
1424 193 : ccuCollService->Init();
1425 193 : aiCpuCollService->Init();
1426 193 : hostCollService->Init();
1427 :
1428 193 : collServices[AcceleratorState::AIV] = ccuCollService; // host 展开，aiv使用
1429 193 : collServices[AcceleratorState::AIV_ONLY] = ccuCollService; // host 展开，aiv使用
1430 193 : collServices[AcceleratorState::CCU_MS] = ccuCollService; // host 展开，ccu使用
1431 193 : collServices[AcceleratorState::CCU_SCHED] = ccuCollService; // host 展开，ccu使用
1432 193 : collServices[AcceleratorState::AICPU_TS] = aiCpuCollService; // aicpu 展开
1433 193 : collServices[AcceleratorState::HOSTCPU_TS] = hostCollService; // host 展开，图模式使用
1434 :
1435 579 : HCCL_INFO("CommunicatorImpl::InitCollServices end");
1436 386 : return;
1437 193 : }
1438 :
1439 41 : HcclResult CommunicatorImpl::InitTraceManager()
1440 : {
1441 : /* 申请trace资源信息 */
1442 41 : std::string logInfo = "HCCL_";
1443 41 : logInfo.append(std::to_string(SalGetTid()));
1444 41 : logInfo.append("_");
1445 41 : logInfo.append(std::to_string(GetDeviceLogicId()));
1446 41 : logInfo.append("_");
1447 41 : logInfo.append(std::to_string(idIndex));
1448 41 : trace = std::make_unique<Trace>();
1449 41 : CHK_PTR_NULL(trace);
1450 41 : CHK_RET(trace->Init(logInfo));
1451 41 : return HCCL_SUCCESS;
1452 41 : }
1453 :
1454 47 : void CommunicatorImpl::InitHDCommunicate()
1455 : {
1456 : // 不管是aicpu还是ccu都初始化
1457 141 : HCCL_INFO("Enter [CommunicatorImpl::InitHDCommunicate]");
1458 47 : kfcControlTransferH2D = std::make_unique<HDCommunicate>(devLogicId, HCCLV2_HDC_TYPE_H2D, sizeof(KfcCommand));
1459 47 : kfcControlTransferH2D->Init();
1460 47 : kfcStatusTransferD2H = std::make_unique<HDCommunicate>(devLogicId, HCCLV2_HDC_TYPE_D2H, sizeof(KfcExecStatus));
1461 47 : kfcStatusTransferD2H->Init();
1462 47 : }
1463 :
1464 50 : void CommunicatorImpl::InitHccpHdc() const
1465 : {
1466 50 : HccpHdcManager::GetInstance().Init(devLogicId);
1467 50 : }
1468 :
1469 1 : void CommunicatorImpl::TryInitCcuFeature() const
1470 : {
1471 1 : TpManager::GetInstance(devLogicId).Init();
1472 3 : HCCL_INFO("[CommunicatorImpl][%s] passed, "
1473 : "will use open source ccu feature.", __func__);
1474 1 : return;
1475 : }
1476 :
1477 52 : void CommunicatorImpl::InitCcuSuperFastLoad()
1478 : {
1479 : //ccu 模式 快速下发模式需要的变量初始化
1480 52 : taskExceptionEnv = EnvConfig::GetInstance().GetLogConfig().GetDfsConfig().taskExceptionEnable;
1481 :
1482 52 : bool hostApiState = ProfilingHandler::GetInstance().GetHostApiState();
1483 52 : bool nodeState = ProfilingHandler::GetInstance().GetHcclNodeState();
1484 52 : bool l0State = ProfilingHandler::GetInstance().GetHcclL0State();
1485 52 : bool l1State = ProfilingHandler::GetInstance().GetHcclL1State();
1486 :
1487 52 : enableProfilingEnv = hostApiState || nodeState || l0State || l1State;
1488 :
1489 156 : HCCL_INFO("taskExceptionEnv[%d], enableProfilingEnv: hostApiState[%d] nodeState[%d] l0State[%d] l1State[%d]",
1490 : taskExceptionEnv, hostApiState, nodeState, l0State, l1State);
1491 52 : }
1492 :
1493 38 : void CommunicatorImpl::InitPreResource()
1494 : {
1495 : // PCIE链路的两端实现enableP2P
1496 38 : auto links = GetFullMeshLinks();
1497 53 : for (auto link : links) {
1498 15 : if (link.GetLinkProtocol() == LinkProtocol::PCIE) {
1499 0 : DeviceId remotePhyId = link.GetRemoteDeviceId();
1500 0 : enableP2PDevices_.push_back(remotePhyId);
1501 : }
1502 : }
1503 38 : CHK_RET_THROW(RuntimeApiException, "EnableP2P Failed", P2PEnableManager::GetInstance().EnableP2P(enableP2PDevices_));
1504 38 : }
1505 :
1506 844 : void CommunicatorImpl::DeInitPreResource()
1507 : {
1508 844 : (void)P2PEnableManager::GetInstance().DisableP2P(devLogicId, enableP2PDevices_);
1509 844 : }
1510 :
1511 242 : void CommunicatorImpl::InitSocketManager()
1512 : {
1513 242 : socketManager = std::make_unique<SocketManager>(*this, myRank, devPhyId, devLogicId);
1514 242 : if (ranktableInfo != nullptr) {
1515 5 : auto rankIpPortMap = ranktableInfo->GetRankDeviceListenPortMap();
1516 5 : rankIpPortMap_ = std::make_shared<decltype(rankIpPortMap)>(std::move(rankIpPortMap));
1517 5 : socketManager->SetDeviceServerListenPortMap(*rankIpPortMap_);
1518 5 : }
1519 242 : }
1520 :
1521 245 : void CommunicatorImpl::InitRmaConnManager()
1522 : {
1523 245 : rmaConnectionManager = std::make_unique<RmaConnManager>(*this);
1524 245 : }
1525 :
1526 252 : void CommunicatorImpl::InitMemTransportManager()
1527 : {
1528 252 : memTransportManager = std::make_unique<MemTransportManager>(*this);
1529 252 : }
1530 :
1531 37 : void CommunicatorImpl::InitHostDeviceSyncNotifyManager()
1532 : {
1533 37 : hostDeviceSyncNotifyManager = std::make_unique<HostDeviceSyncNotifyManager>();
1534 37 : }
1535 :
1536 567 : const string &CommunicatorImpl::GetId() const
1537 : {
1538 567 : return id;
1539 : }
1540 :
1541 20 : u32 CommunicatorImpl::GetIdIndex() const
1542 : {
1543 20 : return idIndex;
1544 : }
1545 :
1546 298 : const string &CommunicatorImpl::GetEstablishLinkSocketTag() const
1547 : {
1548 298 : return establishLinkSocketTag;
1549 : }
1550 :
1551 875 : RankId CommunicatorImpl::GetMyRank() const
1552 : {
1553 875 : return myRank;
1554 : }
1555 :
1556 622 : u32 CommunicatorImpl::GetRankSize() const
1557 : {
1558 622 : return rankSize;
1559 : }
1560 :
1561 591 : u32 CommunicatorImpl::GetDeviceLogicId() const
1562 : {
1563 591 : return devLogicId;
1564 : }
1565 :
1566 34 : u32 CommunicatorImpl::GetDevicePhyId() const
1567 : {
1568 34 : return devPhyId;
1569 : }
1570 :
1571 77 : u64 CommunicatorImpl::GetBufferSize() const
1572 : {
1573 77 : return cclBufferSize;
1574 : }
1575 :
1576 3 : u32 CommunicatorImpl::GetSubmittedOpCnt() const
1577 : {
1578 3 : return submittedOpCnt;
1579 : }
1580 :
1581 0 : u32 CommunicatorImpl::GetOpBaseOpIndex() const
1582 : {
1583 0 : return opBaseOpIndex;
1584 : }
1585 :
1586 12 : u32 CommunicatorImpl::GetOpIndex() const
1587 : {
1588 12 : return opIndex;
1589 : }
1590 :
1591 157 : bool CommunicatorImpl::GetOpAiCpuTSFeatureFlag() const
1592 : {
1593 157 : return opExecuteConfig.accState == AcceleratorState::AICPU_TS;
1594 : }
1595 :
1596 0 : bool CommunicatorImpl::GetCommAiCpuTSFeatureFlag() const
1597 : {
1598 0 : return commExecuteConfig.accState == AcceleratorState::AICPU_TS;
1599 : }
1600 :
1601 235 : const DevType &CommunicatorImpl::GetDevType() const
1602 : {
1603 705 : HCCL_INFO("CommunicatorImpl::DevType is %s", devType.Describe().c_str());
1604 235 : return devType;
1605 : }
1606 :
1607 2507 : shared_ptr<RankGraph> CommunicatorImpl::GetRankGraph() const
1608 : {
1609 7521 : HCCL_INFO("CommunicatorImpl::GetRankGraph ");
1610 2507 : return rankGraph;
1611 : }
1612 :
1613 40 : DataBufManager &CommunicatorImpl::GetDataBufferManager() const
1614 : {
1615 40 : CHECK_NULLPTR(dataBufferManager, "dataBufferManager is nullptr!");
1616 40 : return *dataBufferManager;
1617 : }
1618 :
1619 197 : LocalRmaBufManager &CommunicatorImpl::GetLocalRmaBufManager() const
1620 : {
1621 197 : CHECK_NULLPTR(localRmaBufManager, "localRmaBufManager is nullptr!");
1622 197 : return *localRmaBufManager;
1623 : }
1624 :
1625 0 : RemoteRmaBufManager &CommunicatorImpl::GetRemoteRmaBufManager() const
1626 : {
1627 0 : CHECK_NULLPTR(remoteRmaBufManager, "remoteRmaBufManager is nullptr!");
1628 0 : return *remoteRmaBufManager;
1629 : }
1630 :
1631 5 : QueueNotifyManager &CommunicatorImpl::GetAicpuQueueNotifyManager() const
1632 : {
1633 5 : CHECK_NULLPTR(aicpuQueueNotifyManager_, "aicpuQueueNotifyManager is nullptr!");
1634 5 : return *aicpuQueueNotifyManager_;
1635 : }
1636 :
1637 4 : QueueNotifyManager &CommunicatorImpl::GetCcuQueueNotifyManager() const
1638 : {
1639 4 : CHECK_NULLPTR(ccuQueueNotifyManager_, "ccuQueueNotifyManager is nullptr!");
1640 4 : return *ccuQueueNotifyManager_;
1641 : }
1642 :
1643 1 : ConnLocalNotifyManager &CommunicatorImpl::GetConnLocalNotifyManager() const
1644 : {
1645 1 : CHECK_NULLPTR(connLocalNotifyManager, "connLocalNotifyManager is nullptr!");
1646 1 : return *connLocalNotifyManager;
1647 : }
1648 :
1649 1 : ConnLocalCntNotifyManager &CommunicatorImpl::GetConnLocalCntNotifyManager() const
1650 : {
1651 1 : CHECK_NULLPTR(connLocalCntNotifyManager, "connLocalCntNotifyManager is nullptr!");
1652 1 : return *connLocalCntNotifyManager;
1653 : }
1654 :
1655 13 : QueueWaitGroupCntNotifyManager &CommunicatorImpl::GetQueueWaitGroupCntNotifyManager() const
1656 : {
1657 13 : CHECK_NULLPTR(queueWaitGroupCntNotifyManager, "queueWaitGroupCntNotifyManager is nullptr!");
1658 13 : return *queueWaitGroupCntNotifyManager;
1659 : }
1660 :
1661 13 : QueueBcastPostCntNotifyManager &CommunicatorImpl::GetBcastPostCntNotifyManager() const
1662 : {
1663 13 : CHECK_NULLPTR(queueBcastPostCntNotifyManager, "queueBcastPostCntNotifyManager is nullptr!");
1664 13 : return *queueBcastPostCntNotifyManager;
1665 : }
1666 :
1667 11 : CcuStreamSyncNotifyManager &CommunicatorImpl::GetCcuStreamSyncNotifyManager() const
1668 : {
1669 11 : CHECK_NULLPTR(ccuStreamSyncNotifyManager, "ccuStreamSyncNotifyManager is nullptr!");
1670 11 : return *ccuStreamSyncNotifyManager;
1671 : }
1672 :
1673 133 : StreamManager &CommunicatorImpl::GetStreamManager() const
1674 : {
1675 134 : CHECK_NULLPTR(streamManager, "streamManager is nullptr!");
1676 132 : return *streamManager;
1677 : }
1678 :
1679 33 : AicpuStreamManager &CommunicatorImpl::GetAicpuStreamManager() const
1680 : {
1681 33 : CHECK_NULLPTR(aicpuStreamManager, "aicpuStreamManager is nullptr!");
1682 33 : return *aicpuStreamManager;
1683 : }
1684 :
1685 50 : CollServiceBase *CommunicatorImpl::GetCollService() const
1686 : {
1687 50 : return collService;
1688 : }
1689 :
1690 7 : CollServiceBase *CommunicatorImpl::GetCcuCollService() const
1691 : {
1692 : // 仅在Task Exception下使用，异常捕获由TaskExceptionHandler::Process管理
1693 7 : if (collServices.find(AcceleratorState::CCU_SCHED) != collServices.end()) {
1694 6 : return collServices.at(AcceleratorState::CCU_SCHED).get();
1695 : }
1696 : else {
1697 1 : std::string msg{"[CommunicatorImpl] Communicator uninitialized, this should not be arrived"};
1698 4 : MACRO_THROW(NullPtrException, msg);
1699 1 : }
1700 : }
1701 :
1702 70 : SocketManager &CommunicatorImpl::GetSocketManager() const
1703 : {
1704 70 : CHECK_NULLPTR(socketManager, "socketManager is nullptr!");
1705 70 : return *socketManager;
1706 : }
1707 :
1708 219 : RmaConnManager &CommunicatorImpl::GetRmaConnManager() const
1709 : {
1710 219 : CHECK_NULLPTR(rmaConnectionManager, "rmaConnectionManager is nullptr!");
1711 219 : return *rmaConnectionManager;
1712 : }
1713 :
1714 139 : CollOperator *CommunicatorImpl::GetCurrentCollOperator() const
1715 : {
1716 139 : CHECK_NULLPTR(currentCollOperator, "currentCollOperator is nullptr!");
1717 139 : return currentCollOperator.get();
1718 : }
1719 :
1720 1981264 : MemTransportManager *CommunicatorImpl::GetMemTransportManager() const
1721 : {
1722 1981264 : return memTransportManager.get();
1723 : }
1724 :
1725 84 : bool CommunicatorImpl::GetOpCcuFeatureFlag() const
1726 : {
1727 84 : return IsOpUsingCcuMs() || IsOpUsingCcuSched(); // 算子粒度
1728 : }
1729 :
1730 1 : bool CommunicatorImpl::GetCommCcuFeatureFlag() const
1731 : {
1732 1 : return IsCommUsingCcuMs() || IsCommUsingCcuSched(); // 通信域粒度
1733 : }
1734 :
1735 0 : bool CommunicatorImpl::IsOpUsingAiv() const
1736 : {
1737 0 : return GetOpExecuteConfig().accState == AcceleratorState::AIV;
1738 : }
1739 :
1740 0 : bool CommunicatorImpl::IsOpUsingAivOnly() const
1741 : {
1742 0 : return GetOpExecuteConfig().accState == AcceleratorState::AIV_ONLY;
1743 : }
1744 :
1745 0 : bool CommunicatorImpl::GetOpAivFeatureFlag() const
1746 : {
1747 0 : return IsOpUsingAiv() || IsOpUsingAivOnly();
1748 : }
1749 :
1750 7 : HcclResult CommunicatorImpl::AllocCommResource(void *mc2Tiling, void **commContext)
1751 : {
1752 : try {
1753 7 : AcceleratorState acceleratorState;
1754 22 : CHK_RET(GetTilingAccelerator(mc2Tiling, acceleratorState));
1755 2 : OpExecuteConfig mc2AcceConfig;
1756 2 : mc2AcceConfig.accState = acceleratorState;
1757 2 : SetOpExecuteConfig(mc2AcceConfig);
1758 2 : SelectCollService();
1759 2 : isLoadOp = true;
1760 2 : WaitReady();
1761 0 : collService->AllocCommResource(mc2Tiling, commContext, acceleratorState);
1762 2 : } catch (HcclException &e) {
1763 6 : HCCL_ERROR(e.what());
1764 2 : PrintBackTrace(e);
1765 2 : return e.GetErrorCode();
1766 2 : } catch (exception &e) {
1767 0 : HCCL_ERROR(e.what());
1768 0 : return HcclResult::HCCL_E_INTERNAL;
1769 0 : } catch (...) {
1770 0 : HCCL_ERROR("Unknown error occurs!");
1771 0 : return HcclResult::HCCL_E_INTERNAL;
1772 0 : }
1773 0 : return HcclResult::HCCL_SUCCESS;
1774 : }
1775 :
1776 1 : HcclResult CommunicatorImpl::GetCcuTaskInfo(void *tilingData, void *ccuTaskGroup) const
1777 : {
1778 1 : if (!GetCommCcuFeatureFlag()) { // 通信域粒度
1779 3 : HCCL_ERROR("CommunicatorImpl::GetCcuTaskInfo: ccu is not used, can't GetCcuTaskInfo.");
1780 1 : return HCCL_E_NOT_SUPPORT;
1781 : }
1782 : try {
1783 0 : WaitReady();
1784 0 : collService->GetCcuTaskInfo(tilingData, ccuTaskGroup);
1785 0 : } catch (HcclException &e) {
1786 0 : HCCL_ERROR(e.what());
1787 0 : PrintBackTrace(e);
1788 0 : return e.GetErrorCode();
1789 0 : } catch (exception &e) {
1790 0 : HCCL_ERROR(e.what());
1791 0 : return HcclResult::HCCL_E_INTERNAL;
1792 0 : } catch (...) {
1793 0 : HCCL_ERROR("Unknown error occurs!");
1794 0 : return HcclResult::HCCL_E_INTERNAL;
1795 0 : }
1796 0 : return HcclResult::HCCL_SUCCESS;
1797 : }
1798 :
1799 4 : u32 CommunicatorImpl::GetCcuMc2ServerNum()
1800 : {
1801 7 : if (collServices.find(AcceleratorState::CCU_MS) == collServices.end() ||
1802 7 : collServices.find(AcceleratorState::CCU_SCHED) == collServices.end()) {
1803 1 : THROW<InternalException>("[CommunicatorImpl][%s] not create collServices type "
1804 : "CCU_MS and CCU_SCHED", __func__);
1805 : }
1806 :
1807 3 : auto ccuMc2ServerNum = collServices[AcceleratorState::CCU_MS]->GetCcuMc2ServerNum();
1808 :
1809 3 : return ccuMc2ServerNum;
1810 : }
1811 :
1812 : /* topoDescs 当前只支持l0和l1 */
1813 2 : HcclResult CommunicatorImpl::GetTopoDesc(HcclTopoDescs *topoDescs, uint32_t topoSize) const
1814 : {
1815 2 : if (topoSize < static_cast<uint32_t>(HcclTopoLevel::HCCL_TOPO_MAX)) {
1816 3 : HCCL_ERROR("topoDescs size is not enough, please check topoSize[%u]", topoSize);
1817 1 : return HCCL_E_PARA;
1818 : }
1819 :
1820 1 : topoDescs[static_cast<uint32_t>(HcclTopoLevel::HCCL_TOPO_L0)].algSets = HCCL_ALG_MESH;
1821 1 : topoDescs[static_cast<uint32_t>(HcclTopoLevel::HCCL_TOPO_L1)].algSets = 0;
1822 :
1823 1 : topoDescs[static_cast<uint32_t>(HcclTopoLevel::HCCL_TOPO_L0)].rankSize = rankSize;
1824 1 : topoDescs[static_cast<uint32_t>(HcclTopoLevel::HCCL_TOPO_L1)].rankSize = 0;
1825 :
1826 1 : return HcclResult::HCCL_SUCCESS;
1827 : }
1828 :
1829 16 : HostDeviceSyncNotifyManager &CommunicatorImpl::GetHostDeviceSyncNotifyManager() const
1830 : {
1831 16 : return *hostDeviceSyncNotifyManager;
1832 : }
1833 :
1834 28 : Trace &CommunicatorImpl::GetTrace() const
1835 : {
1836 28 : return *trace;
1837 : }
1838 :
1839 7 : HDCommunicate &CommunicatorImpl::GetKfcControlTransferH2D() const
1840 : {
1841 7 : return *kfcControlTransferH2D;
1842 : }
1843 :
1844 7 : HDCommunicate &CommunicatorImpl::GetKfcStatusTransferD2H() const
1845 : {
1846 7 : return *kfcStatusTransferD2H;
1847 : }
1848 :
1849 : constexpr u32 WAIT_CMD_TIMEOUT = 10 * 1000; // 最大等待10秒
1850 :
1851 3 : HcclResult CommunicatorImpl::Suspend()
1852 : {
1853 15 : TRY_CATCH_RETURN(
1854 : if (isSuspended) {
1855 : HCCL_WARNING("[NsRecovery][Suspend] The current communication has been suspended, no need to suspend again.");
1856 : return HcclResult::HCCL_SUCCESS;
1857 : }
1858 : isSuspended = true;
1859 : if (!isAicpuKernelLaunched) {
1860 : HCCL_INFO("[NsRecovery][Suspend] Aicpu kernel is not launched yet. Suspend host only.");
1861 : return HcclResult::HCCL_SUCCESS;
1862 : }
1863 : KfcCommand opCmd = KfcCommand::NS_STOP_LAUNCH;
1864 : CHK_RET(kfcControlTransferH2D->Put(0, sizeof(KfcCommand), reinterpret_cast<uint8_t *>(&opCmd)));
1865 : HCCL_INFO("[NsRecovery][Suspend] send KfcCommand[%d] success, which is NS_STOP_LAUNCH.", opCmd);
1866 : KfcExecStatus opInfo;
1867 : auto timeout = std::chrono::milliseconds(WAIT_CMD_TIMEOUT);
1868 : auto startTime = std::chrono::steady_clock::now();
1869 : while (true) {
1870 : CHK_RET(kfcStatusTransferD2H->Get(0, sizeof(KfcExecStatus), reinterpret_cast<uint8_t *>(&opInfo)));
1871 : if (opInfo.kfcStatus == KfcStatus::STOP_LAUNCH_DONE) {
1872 : HCCL_INFO("[NsRecovery][Suspend] received KfcStatus[%d], which is STOP_LAUNCH_DONE", opInfo.kfcStatus);
1873 : return HcclResult::HCCL_E_SUSPENDING;
1874 : } else if (opInfo.kfcStatus == KfcStatus::ERROR){
1875 : HCCL_ERROR("[NsRecovery][Suspend] received KfcStatus[%d], which is ERROR", opInfo.kfcStatus);
1876 : return HcclResult::HCCL_E_INTERNAL;
1877 : } else {
1878 : if((std::chrono::steady_clock::now() - startTime) >= timeout){
1879 : HCCL_ERROR("[NsRecovery][Suspend] Wait suspend response status timeout[%u ms] and get the opExecStatus is [%u].", WAIT_CMD_TIMEOUT,
1880 : opInfo.kfcStatus);
1881 : return HcclResult::HCCL_E_TIMEOUT;
1882 : }
1883 : continue;
1884 : }
1885 : }
1886 : );
1887 : return HcclResult::HCCL_SUCCESS;
1888 : }
1889 :
1890 4 : HcclResult CommunicatorImpl::Clean()
1891 : {
1892 28 : TRY_CATCH_RETURN(
1893 : if (!isSuspended) {
1894 : HCCL_ERROR("[NsRecovery][Clean] The current communication is not suspended, cannot clean.");
1895 : return HcclResult::HCCL_E_NOT_SUPPORT;
1896 : }
1897 : isSuspended = true;
1898 : if (isCleaned) {
1899 : HCCL_WARNING("[NsRecovery][Clean] The current communication has been cleaned, no need to clean again.");
1900 : return HcclResult::HCCL_SUCCESS;
1901 : }
1902 : isCleaned = true;
1903 : // 清理host侧资源
1904 : if (GetOpCcuFeatureFlag()) { // 算子粒度加速模式
1905 : if (collService == nullptr) { // 当前通信域没下发过算子
1906 : HCCL_WARNING("[NsRecovery][Clean] The current communication has not loaded op, no need to clean.");
1907 : return HcclResult::HCCL_SUCCESS;
1908 : }
1909 : HCCL_INFO("[NsRecovery][Clean] start to clean host. ccu flag is true");
1910 : auto collServiceCcu = dynamic_cast<CollServiceDeviceMode *>(collService);
1911 : CHECK_NULLPTR(collServiceCcu, "collServiceBase cast to CollServiceDeviceMode failed.");
1912 :
1913 : CcuInsPreprocessor *ccuInsPreprocessor = collServiceCcu->GetCcuInsPreprocessor();
1914 : CHECK_NULLPTR(ccuInsPreprocessor, "ccuInsPreprocessor is nullptr!");
1915 :
1916 : CcuCommunicator *ccuComm = ccuInsPreprocessor->GetCcuComm();
1917 : CHECK_NULLPTR(ccuComm, "ccuComm is nullptr!");
1918 :
1919 : CcuTransportMgr *ccuTransportMgr = ccuComm->GetCcuTransportMgr();
1920 : CHECK_NULLPTR(ccuTransportMgr, "ccuTransportMgr is nullptr!");
1921 : ccuTransportMgr->Clean();
1922 : return HcclResult::HCCL_SUCCESS;
1923 : } else {
1924 : HCCL_INFO("[NsRecovery][Clean] start to clean host. ccu flag is false");
1925 : rmaConnectionManager->Clear();
1926 : memTransportManager->Clear();
1927 : }
1928 : if (!isAicpuKernelLaunched) {
1929 : HCCL_INFO("[NsRecovery][Clean] Aicpu kernel is not launched yet. Clean host only.");
1930 : return HcclResult::HCCL_SUCCESS;
1931 : }
1932 : HCCL_INFO("[NsRecovery][Clean] start to clean device, waiting for device STOP_LAUNCH_DONE");
1933 : KfcExecStatus opInfo;
1934 : CHK_RET(kfcStatusTransferD2H->Get(0, sizeof(KfcExecStatus), reinterpret_cast<uint8_t *>(&opInfo)));
1935 : if (opInfo.kfcStatus == KfcStatus::STOP_LAUNCH_DONE) {
1936 : HCCL_INFO("[NsRecovery][Clean] received KfcStatus[%d], which is STOP_LAUNCH_DONE", opInfo.kfcStatus);
1937 : // 通知背景线程清理device侧资源
1938 : KfcCommand opCmd = KfcCommand::NS_CLEAN;
1939 : CHK_RET(kfcControlTransferH2D->Put(0, sizeof(KfcCommand), reinterpret_cast<uint8_t *>(&opCmd)));
1940 : HCCL_INFO("[NsRecovery][Clean] send KfcCommand [%d] success, which is NS_CLEAN", opCmd);
1941 : // 监听背景线程状态
1942 : auto timeout = std::chrono::milliseconds(WAIT_CMD_TIMEOUT);
1943 : auto startTime = std::chrono::steady_clock::now();
1944 : while (true) {
1945 : CHK_RET(kfcStatusTransferD2H->Get(0, sizeof(KfcExecStatus), reinterpret_cast<uint8_t *>(&opInfo)));
1946 : if (opInfo.kfcStatus == KfcStatus::CLEAN_DONE) {
1947 : HCCL_INFO("[NsRecovery][Clean] received KfcStatus[%d], which is CLEAN_DONE", opInfo.kfcStatus);
1948 : return HcclResult::HCCL_E_SUSPENDING;
1949 : } else if (opInfo.kfcStatus == KfcStatus::ERROR){
1950 : HCCL_ERROR("[NsRecovery][Clean] received KfcStatus[%d], which is ERROR", opInfo.kfcStatus);
1951 : return HcclResult::HCCL_E_INTERNAL;
1952 : } else {
1953 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
1954 : HCCL_ERROR("[NsRecovery][Clean] Wait clean response status timeout[%u ms] and get the opExecStatus is [%u].", WAIT_CMD_TIMEOUT,
1955 : opInfo.kfcStatus);
1956 : return HcclResult::HCCL_E_TIMEOUT;
1957 : }
1958 : continue;
1959 : }
1960 : }
1961 : } else {
1962 : std::string msg = StringFormat("[NsRecovery][Clean] Aicpu kernel is not stopped yet. Cannot clean.");
1963 : THROW<InternalException>(msg);
1964 : return HcclResult::HCCL_E_INTERNAL;
1965 : }
1966 : );
1967 : return HcclResult::HCCL_SUCCESS;
1968 : }
1969 :
1970 3 : HcclResult CommunicatorImpl::Resume()
1971 : {
1972 17 : TRY_CATCH_RETURN(
1973 : if (GetCommStatus() == CommStatus::COMM_ERROR) {
1974 : HCCL_ERROR("[NsRecovery][Resume] Comm has been error, can not resume now!");
1975 : return HcclResult::HCCL_E_INTERNAL;
1976 : }
1977 : if (!isSuspended) {
1978 : HCCL_WARNING("[NsRecovery][Resume] The current communication is normal, no need to resume.");
1979 : return HcclResult::HCCL_SUCCESS;
1980 : }
1981 : if (GetOpCcuFeatureFlag() || GetOpAiCpuTSFeatureFlag() || GetOpAivFeatureFlag()) { // CCU和AICPU AIV // 算子粒度加速模式
1982 : HCCL_INFO("[NsRecovery][Resume] start to Resume.");
1983 : if (collService != nullptr) {
1984 : collService->Resume();
1985 : } else { // 当前通信域没下发过算子
1986 : HCCL_WARNING("[NsRecovery][Resume] The current communication has not loaded op, no need to resume.");
1987 : }
1988 : isSuspended = false;
1989 : isCleaned = false;
1990 : HCCL_INFO("[NsRecovery][Resume] Resume success.");
1991 : } else { // HOST场景不支持
1992 : HCCL_ERROR("[NsRecovery][Resume] HOST is not supported to resume.");
1993 : return HcclResult::HCCL_E_NOT_SUPPORT;
1994 : }
1995 : );
1996 1 : return HcclResult::HCCL_SUCCESS;
1997 : }
1998 :
1999 36 : const NotifyTimeoutCfg &CommunicatorImpl::GetNotifyTimeoutCfg() const
2000 : {
2001 36 : return notifyTimeoutCfg;
2002 : }
2003 :
2004 : /* 当前接口中申请的buffer都是ge图模式下使用 */
2005 9 : HcclResult CommunicatorImpl::CreateCommCclBuf()
2006 : {
2007 27 : HCCL_INFO("[%s] start.", __func__);
2008 9 : if (inCclBuffer == nullptr) {
2009 2 : inCclBuffer = std::make_shared<DevBuffer>(cclBufferSize);
2010 6 : HCCL_INFO("CommunicatorImpl::CreateCommCclBuf, inCclBuffer is %p", inCclBuffer.get());
2011 : }
2012 9 : if (outCclBuffer == nullptr) {
2013 2 : outCclBuffer = std::make_shared<DevBuffer>(cclBufferSize);
2014 6 : HCCL_INFO("CommunicatorImpl::CreateCommCclBuf, outCclBuffer is %p", outCclBuffer.get());
2015 : }
2016 9 : if (indirectInCclBuffer == nullptr) {
2017 2 : indirectInCclBuffer = std::make_shared<DevBuffer>(sizeof(uintptr_t));
2018 6 : HCCL_INFO("Create Indirect In CclBuf success, indirectInCclBuffer = %p", indirectInCclBuffer.get());
2019 : }
2020 9 : if (indirectOutCclBuffer == nullptr) {
2021 2 : indirectOutCclBuffer = std::make_shared<DevBuffer>(sizeof(uintptr_t));
2022 6 : HCCL_INFO("Create Indirect out CclBuf success, indirectOutCclBuffer = %p", indirectOutCclBuffer.get());
2023 : }
2024 9 : return HcclResult::HCCL_SUCCESS;
2025 : }
2026 :
2027 1 : HcclResult CommunicatorImpl::GetInCclBuf(void *&commInputPtr, u64 &commInputSize)
2028 : {
2029 1 : CHK_PTR_NULL(inCclBuffer);
2030 1 : commInputSize = inCclBuffer->GetSize();
2031 1 : commInputPtr = reinterpret_cast<void*>(inCclBuffer->GetAddr());
2032 1 : return HcclResult::HCCL_SUCCESS;
2033 : }
2034 :
2035 1 : HcclResult CommunicatorImpl::GetOutCclBuf(void *&commOutputPtr, u64 &commOutputSize)
2036 : {
2037 1 : CHK_PTR_NULL(outCclBuffer);
2038 1 : commOutputSize = outCclBuffer->GetSize();
2039 1 : commOutputPtr = reinterpret_cast<void*>(outCclBuffer->GetAddr());
2040 1 : return HcclResult::HCCL_SUCCESS;
2041 : }
2042 :
2043 1 : HcclResult CommunicatorImpl::GetIndirectInCclBuf(void *&commIndirectInputPtr, u64 &commIndirectInputSize)
2044 : {
2045 3 : HCCL_INFO("[%s] start.", __func__);
2046 1 : CreateCommCclBuf();
2047 1 : commIndirectInputPtr = reinterpret_cast<void*>(indirectInCclBuffer->GetAddr());
2048 1 : commIndirectInputSize = indirectInCclBuffer->GetSize();
2049 3 : HCCL_INFO("GetIndirectInCclBuf: commIndirectInputPtr[%p], commIndirectInputSize[%lu]", commIndirectInputPtr, commIndirectInputSize);
2050 1 : return HcclResult::HCCL_SUCCESS;
2051 : }
2052 :
2053 1 : HcclResult CommunicatorImpl::GetIndirectOutCclBuf(void *&commIndirectOutputPtr, u64 &commIndirectOutputSize)
2054 : {
2055 3 : HCCL_INFO("[%s] start.", __func__);
2056 1 : CreateCommCclBuf();
2057 1 : commIndirectOutputPtr = reinterpret_cast<void*>(indirectOutCclBuffer->GetAddr());
2058 1 : commIndirectOutputSize = indirectOutCclBuffer->GetSize();
2059 3 : HCCL_INFO("GetIndirectOutCclBuf: commIndirectOutputPtr[%p], commIndirectOutputSize[%lu]", commIndirectOutputPtr, commIndirectOutputSize);
2060 1 : return HcclResult::HCCL_SUCCESS;
2061 : }
2062 :
2063 10 : bool CommunicatorImpl::IsWorldGroup() const
2064 : {
2065 10 : return isWorldGroup;
2066 : }
2067 :
2068 0 : bool CommunicatorImpl::IsCommReady()
2069 : {
2070 0 : CHECK_NULLPTR(collService, "[CommunicatorImpl::IsCommReady] collService is nullptr!");
2071 0 : if (collService->IsAllTransportRecoveredReady(GetId())) {
2072 : // 遗留问题：对Comm状态置为ready
2073 0 : SetCommStatus(CommStatus::COMM_READY);
2074 0 : return true;
2075 : } else {
2076 0 : return false;
2077 : }
2078 : }
2079 :
2080 6 : HcclResult CommunicatorImpl::GetSnapShotDynamicBuf(BinaryStream &buf) const
2081 : {
2082 18 : HCCL_INFO("[CommunicatorImpl][%s] opExecuteConfig.accState is [%u], commExecuteConfig.accState "
2083 : "is [%u], isLoadOp is [%d]",
2084 : __func__, static_cast<u32>(opExecuteConfig.accState), static_cast<u32>(commExecuteConfig.accState),
2085 : isLoadOp);
2086 6 : buf << static_cast<u32>(opExecuteConfig.accState); // 算子粒度 和 通信域粒度 都保存
2087 6 : buf << static_cast<u32>(commExecuteConfig.accState);
2088 6 : buf << isLoadOp;
2089 :
2090 6 : buf << submittedOpCnt;
2091 18 : HCCL_INFO("[CommunicatorImpl][%s], rank[%d], submittedOpCnt[%u]", __func__, myRank, submittedOpCnt);
2092 6 : if (submittedOpCnt == 0) {
2093 5 : return HcclResult::HCCL_SUCCESS;
2094 : }
2095 :
2096 1 : if (currentCollOperator) {
2097 3 : HCCL_INFO("[CommunicatorImpl][%s] opMode is %u", __func__, static_cast<u32>(currentCollOperator->opMode));
2098 1 : buf << static_cast<u32>(currentCollOperator->opMode);
2099 :
2100 3 : HCCL_INFO("[CommunicatorImpl][%s] rank[%d], currentCollOperator", __func__, myRank);
2101 1 : TRY_CATCH_RETURN(collService->GetSnapShotDynamicBuf(*currentCollOperator, buf));
2102 : }
2103 1 : return HcclResult::HCCL_SUCCESS;
2104 : }
2105 :
2106 2 : u32 CommunicatorImpl::GetRanktableCrc(bool isContainLoaId) const
2107 : {
2108 6 : HCCL_INFO("[CommunicatorImpl][%s], rank[%d], id[%s], idIdex[%u]", __func__, myRank, id.c_str(), idIndex);
2109 8 : CHK_PTR_NULL(ranktableInfo);
2110 0 : vector<char> ranktableBuf = ranktableInfo->GetUniqueId(isContainLoaId);
2111 0 : CheckCrc crc;
2112 0 : u32 crcValue = 0;
2113 0 : auto ret = crc.Calc32Crc(reinterpret_cast<const char*>(ranktableBuf.data()), ranktableBuf.size(), &crcValue);
2114 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[CommunicatorImpl][GetRanktableCrc] calculate crc failed, ret[%d]", ret),
2115 : ret);
2116 0 : return crcValue;
2117 0 : }
2118 :
2119 : // 恢复全局通信域
2120 3 : HcclResult CommunicatorImpl::RecoverComm(SnapShotComm &snapShotComm, u32 stepParam, const char *changeInfo)
2121 : {
2122 3 : if (!initFlag) {
2123 2 : initFlag = true;
2124 : try {
2125 6 : HCCL_INFO("[CommunicatorImpl][%s], rank[%d]", __func__, myRank);
2126 : // 将状态设置为resuming
2127 2 : if (GetCommStatus() == CommStatus::COMM_IDLE) {
2128 2 : SetCommStatus(CommStatus::COMM_RESUMING);
2129 : } else {
2130 0 : HCCL_ERROR("Communicator status is not idle, can not resume!");
2131 0 : return HcclResult::HCCL_E_INTERNAL;
2132 : }
2133 2 : RecoverOpMode(snapShotComm.opMode);
2134 2 : InitCommonData(snapShotComm.commParams, snapShotComm.config);
2135 2 : HrtSetDevice(devLogicId);
2136 2 : InitHccpHdc(); // 选择ccu加速模式依赖hdc通道打开ccu驱动
2137 2 : RecoverExeCfgData(snapShotComm.opExecuteConfig, snapShotComm.commExecuteConfig, snapShotComm.isLoadOp); // 算子粒度 和 通信域粒度都恢复
2138 2 : RecoverRankGraphData(snapShotComm, changeInfo);
2139 1 : InitNotifyManager();
2140 1 : InitStreamManager();
2141 1 : InitSocketManager();
2142 1 : InitRmaConnManager();
2143 1 : InitDataBufferManager();
2144 1 : InitMemTransportManager();
2145 1 : InitHostDeviceSyncNotifyManager();
2146 1 : InitUbMemoryTransportMgr();
2147 1 : CollAlgComponentInit();
2148 1 : RegisterAicpuKernel();
2149 1 : InitCollService();
2150 1 : SelectCollService();
2151 1 : InitTraceManager();
2152 1 : DlProfFunction::GetInstance().DlProfFunctionInit();
2153 1 : InitMirrorTaskManager();
2154 1 : CHK_RET(InitProfilingReporter());
2155 1 : InitTaskExceptionHandler();
2156 1 : InitHDCommunicate();
2157 1 : notifyTimeoutCfg.Init();
2158 1 : RecoverTransportData(snapShotComm.submittedOpCnt, snapShotComm.levelRankPairs, stepParam, snapShotComm.linkGroupPair);
2159 1 : } catch (HcclException &e) {
2160 : // 异常时状态返回IDLE
2161 1 : SetCommStatus(CommStatus::COMM_IDLE);
2162 3 : HCCL_ERROR(e.what());
2163 1 : PrintBackTrace(e);
2164 1 : return e.GetErrorCode();
2165 1 : } catch (exception &e) {
2166 : // 异常时状态返回IDLE
2167 0 : SetCommStatus(CommStatus::COMM_IDLE);
2168 0 : HCCL_ERROR(e.what());
2169 0 : return HcclResult::HCCL_E_INTERNAL;
2170 0 : } catch (...) {
2171 : // 异常时状态返回IDLE
2172 0 : SetCommStatus(CommStatus::COMM_IDLE);
2173 0 : HCCL_ERROR("Unknown error occurs!");
2174 0 : return HcclResult::HCCL_E_INTERNAL;
2175 0 : }
2176 1 : return HcclResult::HCCL_SUCCESS;
2177 : }
2178 3 : HCCL_ERROR("[CommunicatorImpl][%s] Repeated calling init method!", __func__);
2179 1 : return HcclResult::HCCL_E_INTERNAL;
2180 : }
2181 :
2182 : // 恢复子通信域
2183 3 : HcclResult CommunicatorImpl::RecoverComm(const SnapShotSubComm &snapShotSubComm, std::unique_ptr<RankGraph> &inputRankGraph, u32 inputStep)
2184 : {
2185 3 : if (!initFlag) {
2186 2 : initFlag = true;
2187 : try {
2188 6 : HCCL_INFO("[CommunicatorImpl][%s], rank[%d]", __func__, myRank);
2189 : // 将状态设置为resuming
2190 2 : if (GetCommStatus() == CommStatus::COMM_IDLE) {
2191 2 : SetCommStatus(CommStatus::COMM_RESUMING);
2192 : } else {
2193 0 : HCCL_ERROR("Communicator status is not idle, can not resume!");
2194 0 : return HcclResult::HCCL_E_INTERNAL;
2195 : }
2196 2 : RecoverOpMode(snapShotSubComm.opMode);
2197 2 : InitCommonDataNotInitDevType(snapShotSubComm.commParams, snapShotSubComm.config);
2198 2 : HrtSetDevice(devLogicId);
2199 2 : InitHccpHdc(); // 选择ccu加速模式依赖hdc通道打开ccu驱动
2200 2 : RecoverExeCfgData(snapShotSubComm.opExecuteConfig, snapShotSubComm.commExecuteConfig, snapShotSubComm.isLoadOp); // 算子粒度 和 通信域粒度都恢复
2201 2 : InitRankGraph(inputRankGraph);
2202 1 : InitNotifyManager();
2203 1 : InitStreamManager();
2204 1 : InitSocketManager();
2205 1 : InitRmaConnManager();
2206 1 : InitDataBufferManager();
2207 1 : InitMemTransportManager();
2208 1 : InitHostDeviceSyncNotifyManager();
2209 1 : InitUbMemoryTransportMgr();
2210 1 : CollAlgComponentInit();
2211 1 : RegisterAicpuKernel();
2212 1 : InitCollService();
2213 1 : SelectCollService();
2214 1 : InitTraceManager();
2215 1 : DlProfFunction::GetInstance().DlProfFunctionInit();
2216 1 : InitMirrorTaskManager();
2217 1 : CHK_RET(InitProfilingReporter());
2218 1 : InitTaskExceptionHandler();
2219 1 : InitHDCommunicate();
2220 1 : RecoverTransportData(snapShotSubComm.submittedOpCnt, snapShotSubComm.levelRankPairs, inputStep, snapShotSubComm.linkGroupPair);
2221 1 : } catch (HcclException &e) {
2222 : // 异常时状态返回IDLE
2223 1 : SetCommStatus(CommStatus::COMM_IDLE);
2224 3 : HCCL_ERROR(e.what());
2225 1 : PrintBackTrace(e);
2226 1 : return e.GetErrorCode();
2227 1 : } catch (exception &e) {
2228 : // 异常时状态返回IDLE
2229 0 : SetCommStatus(CommStatus::COMM_IDLE);
2230 0 : HCCL_ERROR(e.what());
2231 0 : return HcclResult::HCCL_E_INTERNAL;
2232 0 : } catch (...) {
2233 : // 异常时状态返回IDLE
2234 0 : SetCommStatus(CommStatus::COMM_IDLE);
2235 0 : HCCL_ERROR("Unknown error occurs!");
2236 0 : return HcclResult::HCCL_E_INTERNAL;
2237 0 : }
2238 1 : return HcclResult::HCCL_SUCCESS;
2239 : }
2240 3 : HCCL_ERROR("Repeated calling init method!");
2241 1 : return HcclResult::HCCL_E_INTERNAL;
2242 : }
2243 5 : HcclResult CommunicatorImpl::RecoverOpMode(u32 opMode)
2244 : {
2245 5 : if (currentCollOperator == nullptr) {
2246 3 : currentCollOperator = make_unique<CollOperator>();
2247 : }
2248 5 : currentCollOperator->opMode = static_cast<OpMode::Value>(opMode);
2249 5 : return HcclResult::HCCL_SUCCESS;
2250 : }
2251 : // 创建子虚拟拓扑并恢复子通信域
2252 1 : HcclResult CommunicatorImpl::RecoverSubComm(const SnapShotSubComm &snapShotSubComm, CommunicatorImpl *subCommImpl, u32 step)
2253 : {
2254 3 : HCCL_INFO("[CommunicatorImpl][%s] start, myRank is [%d]", __func__, myRank);
2255 1 : vector<u32> rankIds;
2256 4 : for(u32 i = 0; i < snapShotSubComm.rankIds.size(); ++i) {
2257 3 : rankIds.push_back(static_cast<u32>(snapShotSubComm.rankIds[i]));
2258 : }
2259 : try {
2260 1 : if (initFlag) {
2261 : // 创建子虚拟拓扑
2262 0 : std::unique_ptr<RankGraph> subRankGraph = rankGraph->CreateSubRankGraph(rankIds);
2263 : // 初始化子通信域
2264 0 : return subCommImpl->RecoverComm(snapShotSubComm, subRankGraph, step);
2265 0 : } else {
2266 : // 异常时状态返回IDLE
2267 1 : SetCommStatus(CommStatus::COMM_IDLE);
2268 1 : std::string msg = StringFormat("CreateSubComm fail, communicator has not been initialized, please check.");
2269 1 : THROW<InternalException>(msg);
2270 1 : }
2271 1 : } catch (HcclException &e) {
2272 : // 异常时状态返回IDLE
2273 1 : SetCommStatus(CommStatus::COMM_IDLE);
2274 3 : HCCL_ERROR(e.what());
2275 1 : PrintBackTrace(e);
2276 1 : return e.GetErrorCode();
2277 1 : } catch (exception &e) {
2278 : // 异常时状态返回IDLE
2279 0 : SetCommStatus(CommStatus::COMM_IDLE);
2280 0 : HCCL_ERROR(e.what());
2281 0 : return HcclResult::HCCL_E_INTERNAL;
2282 0 : } catch (...) {
2283 : // 异常时状态返回IDLE
2284 0 : SetCommStatus(CommStatus::COMM_IDLE);
2285 0 : HCCL_ERROR("Unknown error occurs!");
2286 0 : return HcclResult::HCCL_E_INTERNAL;
2287 0 : }
2288 : HCCL_ERROR("CreateSubComm fail !");
2289 : return HcclResult::HCCL_E_INTERNAL;
2290 1 : }
2291 : // 恢复全局通信域拓扑信息
2292 2 : HcclResult CommunicatorImpl::RecoverRankGraphData(SnapShotComm &snapShotComm, const char *changeInfo)
2293 : {
2294 6 : HCCL_INFO("[CommunicatorImpl][%s] start, rank[%d]", __func__, myRank);
2295 :
2296 : // 根据changedInfo更新快照信息
2297 2 : auto ret = DiffRankUpdater(changeInfo, snapShotComm.rankTableInfo);
2298 2 : if (ret != HcclResult::HCCL_SUCCESS) {
2299 2 : THROW<InternalException>("DiffRankUpdater failed");
2300 : }
2301 :
2302 0 : RankGraphBuilder rankGraphBuilder;
2303 0 : rankGraph = rankGraphBuilder.RecoverBuild(snapShotComm.rankTableInfo, snapShotComm.topoInfo, myRank);
2304 0 : ranktableInfo = rankGraphBuilder.GetRankTableInfo(); // 获取ranktable信息
2305 0 : HCCL_INFO(
2306 : "[CommunicatorImpl][%s] Recover topo data from snapshot, rank[%d], id[%s], idIndex[%u], RankTableInfo[%s]", __func__,
2307 : myRank, id.c_str(), idIndex, ranktableInfo->Describe().c_str());
2308 0 : topoInfo = rankGraphBuilder.GetTopoInfo(); // 获取topo信息
2309 0 : rankSize = rankGraph->GetRankSize();
2310 :
2311 0 : CheckRankGraph();
2312 0 : HCCL_INFO("Recover topo data from snapshot success.");
2313 0 : return HcclResult::HCCL_SUCCESS;
2314 0 : }
2315 : // 恢复通信域transport信息
2316 0 : HcclResult CommunicatorImpl::RecoverTransportData(u32 savedSubmittedOpCnt, const vector<std::pair<u32, RankId>> &levelRankPairs, u32 savedStep, vector<std::pair<LinkGroup, u32>> linkGroupPair)
2317 : {
2318 0 : HCCL_INFO("[CommunicatorImpl][%s] Recover transport data from snapshot.levelRankPairs size is %u", __func__, levelRankPairs.size());
2319 0 : vector<LinkData> links;
2320 :
2321 0 : for (uint32_t i = 0; i < levelRankPairs.size(); ++i) {
2322 0 : CHK_PTR_NULL(rankGraph);
2323 0 : std::vector<NetInstance::Path> paths = rankGraph->GetPaths(levelRankPairs[i].first, myRank, levelRankPairs[i].second);
2324 0 : for (NetInstance::Path &path : paths) {
2325 0 : links.emplace_back(LinkData(path));
2326 : }
2327 0 : }
2328 : // 指令的下标是指令的个数 - 1
2329 0 : collOpIndex = savedSubmittedOpCnt - 1;
2330 0 : step = savedStep;
2331 : // 建transport
2332 0 : collService->RecoverTransport(links, linkGroupPair);
2333 0 : HCCL_INFO("Recover transport data from snapshot success.");
2334 0 : return HcclResult::HCCL_SUCCESS;
2335 0 : }
2336 :
2337 84 : void CommunicatorImpl::WaitReady() const
2338 : {
2339 84 : constexpr u32 loadWaitTimeOut = 10 * 1000; // 待修改，定义最大等待10秒
2340 84 : auto timeout = std::chrono::milliseconds(loadWaitTimeOut);
2341 :
2342 252 : HCCL_INFO("[CommunicatorImpl][%s] start", __func__);
2343 84 : HcclUs startTime = std::chrono::steady_clock::now();
2344 : while (true) {
2345 398642009 : if (GetCommStatus() == CommStatus::COMM_READY) {
2346 82 : break;
2347 : }
2348 398641927 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
2349 2 : THROW<InternalException>("Wait COMM_READY timeout, commId[%s]", id.c_str());
2350 : }
2351 : }
2352 246 : HCCL_INFO("[CommunicatorImpl][%s] end", __func__);
2353 82 : }
2354 :
2355 5 : u32 CommunicatorImpl::GetCollOpIndex() const
2356 : {
2357 5 : return collOpIndex;
2358 : }
2359 :
2360 5 : u32 CommunicatorImpl::GetStep() const
2361 : {
2362 5 : return step;
2363 : }
2364 :
2365 4 : std::set<RankId> CommunicatorImpl::GetNeighboorRanks() const
2366 : {
2367 4 : return rankGraph->GetNetInstanceByRankId(0,myRank)->GetRankIds();
2368 : }
2369 :
2370 284 : void CommunicatorImpl::InitMirrorTaskManager()
2371 : {
2372 284 : mirrorTaskManager = std::make_unique<MirrorTaskManager>(devLogicId,
2373 284 : &GlobalMirrorTasks::Instance(), false); // host侧写死
2374 284 : }
2375 :
2376 130 : MirrorTaskManager &CommunicatorImpl::GetMirrorTaskManager() const
2377 : {
2378 131 : CHECK_NULLPTR(mirrorTaskManager, "mirrorTaskManager is nullptr!");
2379 129 : return *mirrorTaskManager;
2380 : }
2381 :
2382 846 : CommunicatorImpl::~CommunicatorImpl()
2383 : {
2384 2538 : HCCL_INFO("[~CommunicatorImpl] start CommunicatorImpl destroy, commId[%s]", id.c_str());
2385 846 : (void)DestroyDpuKernelResource();
2386 846 : (void)DestroyDpuTaskexpShmemInDevice();
2387 846 : (void)DestroyKFCWorkSpaceVA();
2388 : // 释放
2389 846 : if (hostShareBuf != nullptr) {
2390 2 : free(hostShareBuf);
2391 2 : hostShareBuf = nullptr;
2392 : }
2393 846 : auto outerIt = g_taskServiceMap.find(id);
2394 846 : if (outerIt != g_taskServiceMap.end()) {
2395 0 : outerIt->second.erase(devLogicId);
2396 0 : if (outerIt->second.empty()) {
2397 0 : g_taskServiceMap.erase(id);
2398 : }
2399 : }
2400 846 : g_taskExpMemMap.erase(id);
2401 846 : (void)NotifyAicpuDestroyComm();
2402 846 : ccuDrvHandle = nullptr;
2403 :
2404 846 : DeInitPreResource();
2405 846 : }
2406 :
2407 846 : HcclResult CommunicatorImpl::DestroyDpuKernelResource()
2408 : {
2409 : // 终止Dpu Kernel的TaskRun
2410 846 : if (!isDpuKernelLaunched) {
2411 835 : return HCCL_SUCCESS;
2412 : }
2413 :
2414 17 : CHK_RET(WaitDpuKernelThreadTerminate());
2415 :
2416 : // 切换回 dpu ctx
2417 9 : aclError aclRet = aclrtSetCurrentContext(dpuContext);
2418 9 : if (ACL_SUCCESS != aclRet) {
2419 0 : HCCL_ERROR("set dpu Ctx Failed, aclReturn[%d]", aclRet);
2420 0 : return HCCL_E_RUNTIME;
2421 : }
2422 : // 销毁局部流
2423 9 : aclRet = aclrtDestroyStreamForce(dpuStream);
2424 9 : if (ACL_SUCCESS != aclRet) {
2425 6 : HCCL_ERROR("Destroy Stream Failed, aclReturn[%d]", aclRet);
2426 2 : aclRet = aclrtSetCurrentContext(npuContext);
2427 8 : CHK_PRT_RET(aclRet == ACL_SUCCESS, HCCL_ERROR("set npu Ctx Failed, aclReturn[%d]", aclRet), HCCL_E_RUNTIME);
2428 0 : return HCCL_E_RUNTIME;
2429 : }
2430 7 : if (g_commNum > 1) {
2431 1 : g_commNum--;
2432 : } else {
2433 : // reset DPU kernel 线程
2434 6 : HcclResult ret = HrtResetXpuDevice(TEMP_DEV_TYPE_DPU, 0);
2435 6 : if (HCCL_SUCCESS != ret) {
2436 18 : HCCL_ERROR("ResetXpuDevice Failed, return[%d]", ret);
2437 6 : aclRet = aclrtSetCurrentContext(npuContext);
2438 24 : CHK_PRT_RET(aclRet == ACL_SUCCESS, HCCL_ERROR("set npu Ctx Failed, aclReturn[%d]", aclRet), HCCL_E_RUNTIME);
2439 0 : return HCCL_E_RUNTIME;
2440 : }
2441 : }
2442 : // 切回 npu ctx
2443 1 : aclRet = aclrtSetCurrentContext(npuContext);
2444 1 : if (ACL_SUCCESS != aclRet) {
2445 0 : HCCL_ERROR("set npu Ctx Failed, aclReturn[%d]", aclRet);
2446 0 : return HCCL_E_RUNTIME;
2447 : }
2448 :
2449 1 : return HCCL_SUCCESS;
2450 : }
2451 :
2452 10 : HcclResult CommunicatorImpl::WaitDpuKernelThreadTerminate()
2453 : {
2454 10 : if (!isDpuKernelLaunched) {
2455 1 : return HCCL_SUCCESS;
2456 : }
2457 :
2458 39 : if (tagDpuShmemArgsMap_[DPUTAG].va_ == nullptr || tagDpuShmemArgsMap_[DPUTAG].accessVA_ == nullptr) {
2459 21 : HCCL_ERROR("[CommunicatorImpl::%s] va_[%p] or accessVA_[%p] is nullptr", __func__, tagDpuShmemArgsMap_[DPUTAG].va_, tagDpuShmemArgsMap_[DPUTAG].accessVA_);
2460 3 : return HCCL_E_MEMORY;
2461 : }
2462 24 : void* hostPtr = tagDpuShmemArgsMap_[DPUTAG].connectType_ == HOST_DEVICE_CONNECT_TYPE_UB ? tagDpuShmemArgsMap_[DPUTAG].va_ : tagDpuShmemArgsMap_[DPUTAG].accessVA_;
2463 : // 先读取是否有dpu线程回写的退出信号
2464 6 : uint8_t flag = 0;
2465 6 : errno_t ret = memcpy_s(&flag, sizeof(flag), hostPtr, sizeof(flag));
2466 6 : if (ret != EOK) {
2467 0 : HCCL_ERROR("Read Terminate TaskRun Signal Fail, return[%d]", ret);
2468 0 : return HCCL_E_INTERNAL;
2469 : }
2470 6 : if (flag == DEVICE_SIGNAL_THIRD) {
2471 5 : return HCCL_SUCCESS; // dpu线程已退出
2472 : }
2473 1 : flag = DEVICE_SIGNAL_SECOND;
2474 1 : ret = memcpy_s(hostPtr, sizeof(flag), &flag, sizeof(flag));
2475 1 : if (ret != EOK) {
2476 0 : HCCL_ERROR("Terminate TaskRun Fail, return[%d]", ret);
2477 0 : return HCCL_E_INTERNAL;
2478 : }
2479 : do {
2480 2541074 : ret = memcpy_s(&flag, sizeof(flag), hostPtr, sizeof(flag));
2481 2541074 : if (ret != EOK) {
2482 0 : HCCL_ERROR("Read Terminate TaskRun Signal Fail, return[%d]", ret);
2483 0 : return HCCL_E_INTERNAL;
2484 : }
2485 2541074 : } while (flag != DEVICE_SIGNAL_THIRD);
2486 1 : return HCCL_SUCCESS;
2487 : }
2488 :
2489 847 : HcclResult CommunicatorImpl::DestroyDpuTaskexpShmemInDevice()
2490 : {
2491 847 : if (!isDpuKernelLaunched) { // 非dpu场景,无共享内存申请
2492 837 : return HCCL_SUCCESS;
2493 : }
2494 :
2495 30 : if (tagDpuShmemArgsMap_[DPUTASKEXCEPTION].va_ == nullptr || tagDpuShmemArgsMap_[DPUTASKEXCEPTION].accessVA_ == nullptr) {
2496 70 : HCCL_ERROR("[CommunicatorImpl::%s] va_[%p] or accessVA_[%p] is nullptr", __func__, tagDpuShmemArgsMap_[DPUTASKEXCEPTION].va_, tagDpuShmemArgsMap_[DPUTASKEXCEPTION].accessVA_);
2497 10 : return HCCL_E_MEMORY;
2498 : }
2499 0 : void* hostPtr = tagDpuShmemArgsMap_[DPUTASKEXCEPTION].connectType_ == HOST_DEVICE_CONNECT_TYPE_UB ? tagDpuShmemArgsMap_[DPUTASKEXCEPTION].va_ : tagDpuShmemArgsMap_[DPUTASKEXCEPTION].accessVA_;
2500 0 : uint8_t flag = 1;
2501 0 : errno_t ret = memcpy_s(hostPtr, sizeof(flag), &flag, sizeof(flag)); // taskexception首字节作为是否停止标志位；
2502 0 : if (ret != EOK) {
2503 0 : HCCL_ERROR("DestroyDpuTaskexpShmemInDevice Fail, return[%d]", ret);
2504 0 : return HCCL_E_INTERNAL;
2505 : }
2506 0 : auto timeout = std::chrono::milliseconds(WAIT_CMD_TIMEOUT);
2507 0 : auto startTime = std::chrono::steady_clock::now();
2508 : do { // 等待aicpu背景线程response
2509 0 : ret = memcpy_s(&flag, sizeof(flag), hostPtr, sizeof(flag));
2510 0 : if (ret != EOK) {
2511 0 : HCCL_ERROR("Read Device Response Fail, return[%d]", ret);
2512 0 : return HCCL_E_INTERNAL;
2513 : }
2514 0 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
2515 0 : HCCL_WARNING("[CommunicatorImpl][DestroyDpuTaskexpShmemInDevice] Wait device destroy response timeout[%u ms].", WAIT_CMD_TIMEOUT);
2516 0 : return HcclResult::HCCL_E_TIMEOUT;
2517 : }
2518 0 : } while (flag != 0);
2519 0 : return HCCL_SUCCESS;
2520 : }
2521 :
2522 846 : HcclResult CommunicatorImpl::NotifyAicpuDestroyComm()
2523 : {
2524 846 : if (!isAicpuKernelLaunched) {
2525 2514 : HCCL_WARNING("[%s] isAicpuKernelLaunched is false", __func__);
2526 838 : return HcclResult::HCCL_SUCCESS;
2527 : }
2528 :
2529 8 : if (kfcControlTransferH2D == nullptr) {
2530 18 : HCCL_WARNING("[%s] kfcControlTransferH2D is null", __func__);
2531 6 : return HcclResult::HCCL_SUCCESS;
2532 : }
2533 :
2534 2 : KfcCommand opCmd = KfcCommand::DESTROY_AICPU_COMM;
2535 6 : HCCL_INFO("[%s] send KfcCommand[%d] begin, which is DESTROY_AICPU_COMM.", __func__, opCmd);
2536 2 : CHK_RET(kfcControlTransferH2D->Put(0, sizeof(KfcCommand), reinterpret_cast<uint8_t *>(&opCmd)));
2537 6 : HCCL_INFO("[%s] send KfcCommand[%d] success, which is DESTROY_AICPU_COMM.", __func__, opCmd);
2538 2 : KfcExecStatus opInfo;
2539 2 : auto timeout = std::chrono::milliseconds(WAIT_CMD_TIMEOUT);
2540 2 : auto startTime = std::chrono::steady_clock::now();
2541 : while (true) {
2542 4606260 : CHK_RET(kfcStatusTransferD2H->Get(0, sizeof(KfcExecStatus), reinterpret_cast<uint8_t *>(&opInfo)));
2543 4606260 : if (opInfo.kfcStatus == KfcStatus::DESTROY_AICPU_COMM_DONE) {
2544 0 : HCCL_INFO("[%s] get KfcStatus[%d], which is DESTROY_AICPU_COMM_DONE", __func__, opInfo.kfcStatus);
2545 0 : return HcclResult::HCCL_SUCCESS;
2546 : }
2547 4606260 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
2548 6 : HCCL_WARNING("[%s] Wait suspend response status timeout[%u ms] and get the "
2549 : "opExecStatus is [%u].", __func__,
2550 : WAIT_CMD_TIMEOUT, opInfo.kfcStatus);
2551 2 : return HcclResult::HCCL_E_TIMEOUT;
2552 : }
2553 4606258 : }
2554 :
2555 : return HcclResult::HCCL_SUCCESS;
2556 : }
2557 :
2558 257 : HcclResult CommunicatorImpl::InitProfilingReporter()
2559 : {
2560 514 : profilingReporter = std::make_unique<ProfilingReporter>(mirrorTaskManager.get(),
2561 514 : &ProfilingHandler::GetInstance());
2562 257 : CHK_RET(ProfilingHandler::GetInstance().Init());
2563 257 : CHK_RET(profilingReporter->Init());
2564 257 : return HCCL_SUCCESS;
2565 : }
2566 :
2567 0 : ProfilingReporter &CommunicatorImpl::GetProfilingReporter() const
2568 : {
2569 0 : CHECK_NULLPTR(profilingReporter, "profilingReporter is nullptr!");
2570 0 : return *profilingReporter;
2571 : }
2572 :
2573 8 : HcclResult CommunicatorImpl::GetOneSidedService(HcclOneSidedService** service) const
2574 : {
2575 8 : CHECK_NULLPTR(oneSidedService, "oneSidedService is nullptr!");
2576 8 : *service = oneSidedService.get();
2577 8 : return HCCL_SUCCESS;
2578 : }
2579 :
2580 8 : void CommunicatorImpl::UpdateProfStat()
2581 : {
2582 8 : profilingReporter->UpdateProfStat();
2583 8 : }
2584 :
2585 6 : void CommunicatorImpl::ReportProfInfo(uint64_t beginTime, bool cachedReq, bool opbased)
2586 : {
2587 : // 上报task信息
2588 6 : profilingReporter->ReportAllTasks(cachedReq);
2589 :
2590 : // 上报opInfo信息
2591 6 : profilingReporter->ReportOp(beginTime, cachedReq, opbased);
2592 6 : }
2593 :
2594 41 : void CommunicatorImpl::InitTaskExceptionHandler() const
2595 : {
2596 41 : TaskExceptionHandler* handler = TaskExceptionHandlerManager::GetHandler(static_cast<size_t>(devLogicId));
2597 41 : CHECK_NULLPTR(handler, "handler is nullptr!");
2598 41 : }
2599 :
2600 39 : void CommunicatorImpl::InitOneSidedService()
2601 : {
2602 117 : HCCL_INFO("[CommunicatorImpl][InitOneSidedService] start!");
2603 39 : oneSidedService = std::make_unique<HcclOneSidedService>(*this);
2604 117 : HCCL_INFO("[CommunicatorImpl][InitOneSidedService] end!");
2605 39 : }
2606 :
2607 2 : u32 CommunicatorImpl::GetUsedChannelCount(u32 dieId)
2608 : {
2609 2 : CHECK_NULLPTR(collService, "collService is nullptr!");
2610 2 : if (!GetOpCcuFeatureFlag()) { return 0; } // 防止非ccu模式进入
2611 2 : CHECK_NULLPTR(dynamic_cast<CollServiceDeviceMode *>(collService), "CollServiceDeviceMode is nullptr!");
2612 2 : CcuJettyMgr *ccuJettyMgr = dynamic_cast<CollServiceDeviceMode *>(collService)
2613 : ->GetCcuInsPreprocessor()
2614 : ->GetCcuComm()
2615 4 : ->GetCcuJettyMgr();
2616 :
2617 2 : if (ccuJettyMgr == nullptr) {
2618 3 : HCCL_WARNING("[CommunicatorImpl][%s] failed, ccuJettyMgr is nullptr, commId[%s].",
2619 : __func__, id.c_str());
2620 1 : return 0;
2621 : }
2622 1 : return ccuJettyMgr->GetUsedChannelCount(dieId);
2623 : }
2624 :
2625 13 : void CommunicatorImpl::RegisterPrintChannelInfoCallback(std::function<void()> callback)
2626 : {
2627 13 : printChannelInfoCallback = callback;
2628 13 : }
2629 :
2630 3 : void CommunicatorImpl::PrintChannelInfoCallback() const
2631 : {
2632 3 : if (printChannelInfoCallback == nullptr) {
2633 6 : HCCL_WARNING("[CommunicatorImpl][PrintChannelInfoCallback] commId[%s], callback function not registered.", id.c_str());
2634 2 : return;
2635 : }
2636 : // ccu建链时channel资源不足，调用回调函数做维测打印
2637 1 : printChannelInfoCallback();
2638 : }
2639 :
2640 296 : void CommunicatorImpl::SetCommStatus(CommStatus commStatus)
2641 : {
2642 296 : status_.store(commStatus);
2643 296 : }
2644 :
2645 398642131 : CommStatus CommunicatorImpl::GetCommStatus() const
2646 : {
2647 398642131 : return status_.load();
2648 : }
2649 :
2650 : std::map<HcclAccelerator, AcceleratorState> accStateMap = {
2651 : {HcclAccelerator::AICPU, AcceleratorState::AICPU_TS},
2652 : {HcclAccelerator::AICPU_TS, AcceleratorState::AICPU_TS},
2653 : {HcclAccelerator::CCU_SCHED, AcceleratorState::CCU_SCHED},
2654 : {HcclAccelerator::DEFAULT, AcceleratorState::CCU_SCHED},
2655 : {HcclAccelerator::CCU_MS, AcceleratorState::CCU_MS}
2656 : };
2657 :
2658 : // 初始化 算子粒度 = 通信域粒度 选择用 算子粒度 ok
2659 26 : void CommunicatorImpl::ExecAlgSelect(const CollOpParams &opParams, const OpMode &opMode)
2660 : {
2661 78 : HCCL_INFO("[CommunicatorImpl][%s] opType[%s], opMode[%s], primary accelerator[%s]", __func__, opParams.opType.Describe().c_str(),
2662 : opMode.Describe().c_str(), opExecuteConfig.accState.Describe().c_str());
2663 : // 调用算法选择接口，获取algName、展开方式、执行方式
2664 26 : CollAlgParams params;
2665 26 : params.opMode = opMode;
2666 26 : params.maxTmpMemSize = GetBufferSize();
2667 26 : params.isMc2 = opParams.isMc2;
2668 26 : if (opParams.isMc2) {
2669 0 : if(accStateMap.find(opParams.commEngine) == accStateMap.end()) {
2670 0 : THROW<NotSupportException>("[CommunicatorImpl][ExecAlgSelect] not support commEngine type[%s]!", opParams.commEngine.Describe().c_str());
2671 : }
2672 0 : opExecuteConfig.accState = accStateMap.find(opParams.commEngine)->second;
2673 : }
2674 26 : OpExecuteConfig inOpExecuteConfig = opExecuteConfig;
2675 26 : params.opExecuteConfig = inOpExecuteConfig;
2676 26 : params.algConfig = opParams.algConfig;
2677 :
2678 78 : HCCL_DEBUG("CommunicatorImpl::ExecAlgSelect currentCollOperator dataType[%s]", currentCollOperator->dataType.Describe().c_str());
2679 26 : auto ret = collAlgComponent->ExecAlgSelect(*currentCollOperator, params, curAlgName, inOpExecuteConfig);
2680 26 : if (ret != HcclResult::HCCL_SUCCESS) {
2681 : std::vector<HcclAlgoType> algos
2682 1 : = std::vector<HcclAlgoType>(HCCL_ALGO_LEVEL_NUM, HcclAlgoType::HCCL_ALGO_TYPE_DEFAULT);
2683 1 : auto configAlgMap = EnvConfig::GetInstance().GetAlgoConfig().GetAlgoConfig();
2684 1 : auto it = configAlgMap.find(opParams.opType);
2685 1 : if (it != configAlgMap.end()) {
2686 0 : algos = it->second;
2687 : }
2688 1 : auto dataSize = opParams.count * DataTypeSizeGet(opParams.dataType);
2689 6 : THROW<NotSupportException>(
2690 : "[CommunicatorImpl][ExecAlgSelect] failed. Error code :%u, opType[%s], opMode[%s], accState[%s], "
2691 : "dataType[%s], reduceOp[%s]. Current algName[%s],algos[0]:[%u],algos[1]:[%u],algos[2]:[%u],algos[3]:[%u], dataSize[%u Bytes] .",
2692 5 : ret, opParams.opType.Describe().c_str(), opMode.Describe().c_str(),
2693 5 : opExecuteConfig.accState.Describe().c_str(), opParams.dataType.Describe().c_str(),
2694 4 : opParams.reduceOp.Describe().c_str(), curAlgName.c_str(), algos[0], algos[1], algos[2], algos[3], dataSize);
2695 2 : }
2696 25 : if(params.isMc2 && (opExecuteConfig.accState == AcceleratorState::CCU_SCHED || opExecuteConfig.accState == AcceleratorState::CCU_MS)) {
2697 0 : algorithmType_ = collAlgComponent->GetAlgorithmTypeForMC2CCU(curAlgName);
2698 : }
2699 25 : auto opAcceStateCacheIt = opAcceStateCache.find({opParams.opType, curAlgName});
2700 25 : if (opAcceStateCacheIt != opAcceStateCache.end()) {
2701 0 : HCCL_INFO("[CommunicatorImpl][%s] opAcceStateCache find, reset accelerator[%s], algName[%s]", __func__, opAcceStateCacheIt->second.first.Describe().c_str(), opAcceStateCacheIt->second.second.c_str());
2702 0 : opExecuteConfig.accState = opAcceStateCacheIt->second.first;
2703 0 : curAlgName = opAcceStateCacheIt->second.second;
2704 0 : ExecAlgSelect(opParams, opMode); // 重新走算法选择(数据量、数据类型、reduce类型不一样，算法可能不一样)
2705 0 : return;
2706 : }
2707 25 : SetOpExecuteConfig(inOpExecuteConfig); // 算子粒度 ok
2708 75 : HCCL_INFO("[CommunicatorImpl][%s] current accelerator[%s], algName[%s], algorithmType[%u]", __func__,
2709 : opExecuteConfig.accState.Describe().c_str(), curAlgName.c_str(), algorithmType_);
2710 25 : SelectCollService();
2711 26 : }
2712 :
2713 57 : void CommunicatorImpl::SelectCollService()
2714 : {
2715 : // 根据执行方式和展开方式，选择对应的CollService
2716 57 : auto mapIt = collServices.find(GetOpExecuteConfig().accState); // 算子粒度
2717 57 : if (mapIt == collServices.end()) {
2718 : auto msg = StringFormat("[CommunicatorImpl][%s] not support, accelerator is %s", __func__,
2719 3 : GetOpExecuteConfig().accState.Describe().c_str());
2720 3 : THROW<NotSupportException>(msg);
2721 3 : }
2722 54 : collService = mapIt->second.get();
2723 54 : }
2724 :
2725 226 : void CommunicatorImpl::CollAlgComponentInit()
2726 : {
2727 226 : HcclMainboardId hcclMainboardId;
2728 226 : HrtGetMainboardId(devLogicId, hcclMainboardId);
2729 226 : CollAlgComponentBuilder collAlgComponentBuilder;
2730 226 : collAlgComponent = collAlgComponentBuilder.SetRankGraph(GetRankGraph().get())
2731 226 : .SetDevType(GetDevType())
2732 226 : .SetMyRank(GetMyRank())
2733 226 : .SetRankSize(GetRankSize())
2734 226 : .SetDmaMode(DmaMode::PUT)
2735 226 : .SetMainboardId(static_cast<uint8_t>(hcclMainboardId))
2736 678 : .EnableDetour(EnvConfig::GetInstance().GetDetourConfig().GetDetourType()
2737 226 : == HcclDetourType::HCCL_DETOUR_ENABLE_2P) // 当前仅支持2P绕路
2738 226 : .Build();
2739 226 : if (collAlgComponent == nullptr) {
2740 0 : HCCL_ERROR("collAlgComponent is a null pointer!");
2741 0 : throw NullPtrException("collAlgComponent is a null pointer!");
2742 : }
2743 678 : HCCL_INFO("[CommunicatorImpl][%s] finished initializing collAlgComponent.", __func__);
2744 226 : }
2745 :
2746 16 : HcclResult CommunicatorImpl::SetAccelerator(HcclAccelerator hcclAccelerator, bool isCcuMsAvailable)
2747 : {
2748 16 : if (isLoadOp) {
2749 : // 已下发过算子，不允许再设置accelerator
2750 3 : HCCL_ERROR("[CommunicatorImpl]SetAccelerator is not allowed after load op.");
2751 1 : return HCCL_E_NOT_SUPPORT;
2752 : }
2753 15 : AcceleratorState commAccelerator;
2754 15 : if (hcclAccelerator == HcclAccelerator::DEFAULT) { // 用户没有配，读环境变量
2755 14 : hcclAccelerator = EnvConfig::GetInstance().GetAlgoConfig().GetHcclAccelerator();
2756 42 : HCCL_RUN_INFO("[CommunicatorImpl][%s] env OpExpansionMode is [%s]", __func__, hcclAccelerator.Describe().c_str());
2757 : }
2758 15 : HcclMainboardId hcclMainboardId;
2759 15 : CHK_RET(HrtGetMainboardId(devLogicId, hcclMainboardId));
2760 :
2761 : // 开启新流程时，仅mc2场景走回legacy通信域，此时不允许使用ms模式
2762 45 : HCCL_WARNING("[CommunicatorImpl][%s] legacy communicator not support ccu ms mode for mc2.",
2763 : __func__);
2764 15 : isCcuMsAvailable = false;
2765 :
2766 15 : switch (hcclAccelerator) {
2767 1 : case HcclAccelerator::CCU_MS:
2768 1 : if (hcclMainboardId == HcclMainboardId::MAINBOARD_PCIE_STD) { // 标卡环境下配置CCU_MS加速模式拦截报错
2769 3 : HCCL_ERROR("[SetAccelerator] hcclAccelerator[%s] not support in %s", hcclAccelerator.Describe().c_str(), hcclMainboardId.Describe().c_str());
2770 1 : return HCCL_E_NOT_SUPPORT;
2771 : }
2772 0 : commAccelerator = isCcuMsAvailable ? AcceleratorState::CCU_MS : AcceleratorState::CCU_SCHED;
2773 0 : break;
2774 0 : case HcclAccelerator::CCU_SCHED:
2775 0 : commAccelerator = AcceleratorState::CCU_SCHED;
2776 0 : if (IsCommWithPCIEProtocol()) {
2777 : // 若当前通信域存在PCIE链路,不支持ccu展开,默认切换为aicpu展开,在大于8卡不支持aicpu场景由后续算法选择部分切换至aiv展开
2778 0 : commAccelerator = AcceleratorState::AICPU_TS;
2779 : }
2780 0 : break;
2781 0 : case HcclAccelerator::AIV:
2782 0 : commAccelerator = AcceleratorState::AIV;
2783 0 : break;
2784 0 : case HcclAccelerator::AIV_ONLY:
2785 0 : commAccelerator = AcceleratorState::AIV_ONLY;
2786 0 : break;
2787 14 : case HcclAccelerator::AICPU_TS:
2788 14 : commAccelerator = AcceleratorState::AICPU_TS;
2789 14 : break;
2790 0 : case HcclAccelerator::HOSTCPU_TS: // 950不支持HOST展开，进行拦截
2791 0 : HCCL_ERROR("[SetAccelerator] hcclAccelerator[%s] not support in 950", hcclAccelerator.Describe().c_str());
2792 0 : return HCCL_E_NOT_SUPPORT;
2793 0 : case HcclAccelerator::AICPU:
2794 0 : HCCL_ERROR("[SetAccelerator] hcclAccelerator[%s] not support", hcclAccelerator.Describe().c_str());
2795 0 : return HCCL_E_NOT_SUPPORT;
2796 0 : default:
2797 0 : HCCL_ERROR("[SetAccelerator] hcclAccelerator[%s] internal error", hcclAccelerator.Describe().c_str());
2798 0 : return HCCL_E_INTERNAL;
2799 : }
2800 14 : OpExecuteConfig inCommExecuteConfig;
2801 14 : inCommExecuteConfig.accState = commAccelerator;
2802 42 : HCCL_DEBUG("[CommunicatorImpl][%s] inCommExecuteConfig[%s]", __func__, inCommExecuteConfig.accState.Describe().c_str());
2803 14 : TRY_CATCH_RETURN(SetCommExecuteConfig(inCommExecuteConfig)); // 设置通信域粒度加速模式，ccu模式需打开ccu驱动
2804 14 : SetOpExecuteConfig(inCommExecuteConfig); // 算子粒度加速模式 同步为 通信域粒度加速模式
2805 42 : HCCL_DEBUG("[CommunicatorImpl][%s] comm accelerator [%s], isCcuMsAvailable is [%d]", __func__, GetCommExecuteConfig().accState.Describe().c_str(), isCcuMsAvailable);
2806 14 : return HCCL_SUCCESS;
2807 : }
2808 :
2809 0 : bool CommunicatorImpl::IsCommWithPCIEProtocol() const
2810 : {
2811 0 : auto links = GetFullMeshLinks();
2812 0 : for (auto link : links) {
2813 0 : if (link.GetLinkProtocol() == LinkProtocol::PCIE) {
2814 0 : HCCL_INFO("[CommunicatorImpl][%s]the current communicator has PCIE link", __func__);
2815 0 : return true;
2816 : }
2817 : }
2818 0 : HCCL_INFO("[CommunicatorImpl][%s]the current communicator does not have a PCIE link", __func__);
2819 0 : return false;
2820 0 : }
2821 :
2822 0 : HcclResult CommunicatorImpl::GetAccelerator(int32_t *accelerator) const
2823 : {
2824 0 : HcclAccelerator hcclAccelerator{HcclAccelerator::DEFAULT};
2825 0 : auto commAccelerator = GetCommExecuteConfig().accState;
2826 0 : std::string acceleraToStr = AcceleratorStateToString.at(commAccelerator);
2827 0 : HCCL_INFO("[CommunicatorImpl][%s] commId[%s], commAccelerator[%s]", __func__, GetId().c_str(),
2828 : acceleraToStr.c_str());
2829 :
2830 0 : switch (commAccelerator) {
2831 0 : case AcceleratorState::CCU_MS:
2832 0 : hcclAccelerator = HcclAccelerator::CCU_MS;
2833 0 : break;
2834 0 : case AcceleratorState::CCU_SCHED:
2835 0 : hcclAccelerator = HcclAccelerator::CCU_SCHED;
2836 0 : break;
2837 0 : case AcceleratorState::AIV:
2838 0 : hcclAccelerator = HcclAccelerator::AIV;
2839 0 : break;
2840 0 : case AcceleratorState::AIV_ONLY:
2841 0 : hcclAccelerator = HcclAccelerator::AIV_ONLY;
2842 0 : break;
2843 0 : case AcceleratorState::AICPU_TS:
2844 0 : hcclAccelerator = HcclAccelerator::AICPU_TS;
2845 0 : break;
2846 0 : case AcceleratorState::HOSTCPU_TS:
2847 0 : hcclAccelerator = HcclAccelerator::HOSTCPU_TS;
2848 0 : break;
2849 0 : case AcceleratorState::AICPU:
2850 0 : hcclAccelerator = HcclAccelerator::AICPU;
2851 0 : break;
2852 0 : default:
2853 0 : HCCL_ERROR("[GetAccelerator] commAccelerator[%s] internal error", acceleraToStr.c_str());
2854 0 : return HCCL_E_INTERNAL;
2855 : }
2856 0 : *accelerator = static_cast<int32_t>(hcclAccelerator);
2857 0 : return HCCL_SUCCESS;
2858 0 : }
2859 :
2860 186 : bool CommunicatorImpl::IsOpUsingCcuMs() const
2861 : {
2862 186 : return GetOpExecuteConfig().accState == AcceleratorState::CCU_MS;
2863 : }
2864 :
2865 168 : bool CommunicatorImpl::IsOpUsingCcuSched() const
2866 : {
2867 168 : return GetOpExecuteConfig().accState == AcceleratorState::CCU_SCHED;
2868 : }
2869 :
2870 4 : bool CommunicatorImpl::IsCommUsingCcuMs() const
2871 : {
2872 4 : return GetCommExecuteConfig().accState == AcceleratorState::CCU_MS;
2873 : }
2874 :
2875 4 : bool CommunicatorImpl::IsCommUsingCcuSched() const
2876 : {
2877 4 : return GetCommExecuteConfig().accState == AcceleratorState::CCU_SCHED;
2878 : }
2879 :
2880 4 : HcclResult CommunicatorImpl::RecoverExeCfgData(const OpExecuteConfig &inOpExeCfg, const OpExecuteConfig &inCommExeCfg, bool inIsLoadOp)
2881 : {
2882 : // mc2目前没有快照恢复，如果增加需要调用该接口
2883 12 : HCCL_INFO("CommunicatorImpl[%s] Recover ExecuteConfig, opAcceState is %s, commAcceState is %s, isLoadOp is %d", __func__,
2884 : inOpExeCfg.accState.Describe().c_str(), inCommExeCfg.accState.Describe().c_str(), inIsLoadOp);
2885 :
2886 : // 恢复加速器类型
2887 4 : SetOpExecuteConfig(inOpExeCfg); // 算子粒度 和 通信域粒度 都考虑
2888 4 : SetCommExecuteConfig(inCommExeCfg);
2889 4 : isLoadOp = inIsLoadOp;
2890 :
2891 12 : HCCL_INFO("Recover OpExecuteConfig data from snapshot success.");
2892 4 : return HcclResult::HCCL_SUCCESS;
2893 : }
2894 :
2895 188 : void CommunicatorImpl::RegisterAcceStateCallBack(std::function<HcclResult(const std::string &commId, bool isUsingCcuMs, bool isUsingCcuSched)> inCallback)
2896 : {
2897 188 : callback = inCallback;
2898 188 : }
2899 :
2900 102 : void CommunicatorImpl::SetOpExecuteConfig(const OpExecuteConfig &inConfig)
2901 : {
2902 102 : opExecuteConfig = inConfig;
2903 306 : HCCL_DEBUG(
2904 : "[CommunicatorImpl][%s] comm id [%s], IsOpUsingCcuMs [%d], IsOpUsingCcuSched [%d]",
2905 : __func__, GetId().c_str(), IsOpUsingCcuMs(), IsOpUsingCcuSched()); // 算子粒度
2906 102 : }
2907 :
2908 1 : void CommunicatorImpl::SetCommExecuteConfig(const OpExecuteConfig& inConfig)
2909 : {
2910 1 : commExecuteConfig = inConfig;
2911 3 : HCCL_DEBUG(
2912 : "[CommunicatorImpl][%s] update comm manager ccu status, comm id [%s], IsCommUsingCcuMs [%d], IsCommUsingCcuSched [%d]",
2913 : __func__, GetId().c_str(), IsCommUsingCcuMs(), IsCommUsingCcuSched()); // 通信域粒度
2914 :
2915 1 : TryInitCcuFeature(); // 单例结构整改前临时方案
2916 :
2917 1 : callback(GetId(), IsCommUsingCcuMs(), IsCommUsingCcuSched());
2918 1 : }
2919 :
2920 2 : HcclResult CommunicatorImpl::CalcTaskNum(OpType opType, DataType dataType, u64 count, u32 &taskNum) const
2921 : {
2922 6 : HCCL_INFO("[CommunicatorImpl][CalcTaskNum] start!");
2923 2 : return collAlgComponent->CalcTaskNum(opType, dataType, count, taskNum);
2924 : }
2925 :
2926 194 : void CommunicatorImpl::InitUbMemoryTransportMgr()
2927 : {
2928 194 : ubMemoryTransportMgr = std::make_unique<UbMemoryTransportMgr>(*this);
2929 194 : }
2930 :
2931 11 : UbMemoryTransportMgr *CommunicatorImpl::GetUbMemoryTransportMgr() const
2932 : {
2933 11 : return ubMemoryTransportMgr.get();
2934 : }
2935 :
2936 58 : HcclResult CommunicatorImpl::HcomSelectAlg(const CollOpParams& opParams, int32_t aivCoreLimit, bool &ifAiv, std::string &algName)
2937 : {
2938 174 : HCCL_INFO("CommunicatorImpl::HcomSelectAlg opType[%s], count[%llu], dataType[%s], HcclReduceOp[%s], aivCoreLimit[%d]",
2939 : opParams.opType.Describe().c_str(), opParams.count, opParams.dataType.Describe().c_str(), opParams.reduceOp.Describe().c_str(), aivCoreLimit);
2940 :
2941 58 : if (GetCommStatus() == CommStatus::COMM_ERROR) {
2942 3 : HCCL_ERROR("Comm has been error, can not select alg now!");
2943 1 : return HcclResult::HCCL_E_INTERNAL;
2944 : }
2945 :
2946 57 : if (isSuspended) {
2947 3 : HCCL_ERROR("Comm has been suspended, can not select alg now!");
2948 1 : return HcclResult::HCCL_E_SUSPENDING;
2949 : }
2950 : // 等待通信域状态为Ready，执行算子下发
2951 56 : WaitReady();
2952 :
2953 56 : std::string tag = ""; // 算法选择不需要传入tag，获取kernel arg的时候会用到
2954 56 : CovertToCurrentCollOperator(tag, opParams, OpMode::OFFLOAD, true, true);
2955 : // 图模式算子加载选择CollService
2956 56 : opExecuteConfig = commExecuteConfig;
2957 56 : ExecAlgSelect(opParams, OpMode::OFFLOAD);
2958 56 : ifAiv = (opExecuteConfig.accState == AcceleratorState::AIV || opExecuteConfig.accState == AcceleratorState::AIV_ONLY);
2959 56 : HcclResult dataTypeChkRes = OpParamsChecker::CheckOpDataTypeOffload(opParams, GetOpCcuFeatureFlag(),
2960 56 : GetOpAiCpuTSFeatureFlag(), ifAiv);
2961 56 : if (dataTypeChkRes != HcclResult::HCCL_SUCCESS) {
2962 0 : HCCL_ERROR("[CommunicatorImpl::HcomSelectAlg] DataType check fail.");
2963 0 : SetCommStatus(CommStatus::COMM_READY);
2964 0 : return dataTypeChkRes;
2965 : }
2966 56 : algName = curAlgName;
2967 :
2968 56 : return HcclResult::HCCL_SUCCESS;
2969 56 : }
2970 :
2971 2 : void CommunicatorImpl::ReportHcclMC2Info(const Stream &kfcStream, Stream &stream, const std::vector<Stream*> &aicpuStreams)
2972 : {
2973 2 : HcclResult hcclRet = InitProfilingReporter();
2974 2 : if (hcclRet != HCCL_SUCCESS) {
2975 0 : HCCL_ERROR("[ReportHcclMC2Info]InitProfilingReporter failed, ret[%d]", static_cast<s32>(hcclRet));
2976 0 : return;
2977 : }
2978 2 : profilingReporter->CallReportMc2CommInfo(kfcStream, stream, aicpuStreams, id, myRank, rankSize, rankInParentComm);
2979 : }
2980 :
2981 4 : void CommunicatorImpl::OpAcceleratorStateFallback()
2982 : {
2983 4 : OpExecuteConfig inOpExecuteConfig;
2984 : // 只要ccu出问题，直接回退到CCU_FALLBACK，走AICPU
2985 4 : switch (opExecuteConfig.accState) {
2986 2 : case AcceleratorState::CCU_MS:
2987 2 : inOpExecuteConfig.accState = AcceleratorState::CCU_FALLBACK;
2988 2 : break;
2989 1 : case AcceleratorState::CCU_SCHED:
2990 1 : inOpExecuteConfig.accState = AcceleratorState::CCU_FALLBACK;
2991 1 : break;
2992 1 : default:
2993 1 : THROW<NotSupportException>(
2994 3 : StringFormat("[CommunicatorImpl::%s] Only supports CCU accelerator rollback", __func__));
2995 : break;
2996 : }
2997 3 : SetOpExecuteConfig(inOpExecuteConfig);
2998 3 : }
2999 :
3000 2 : HcclResult CommunicatorImpl::AcceleratorFallback()
3001 : {
3002 6 : HCCL_RUN_INFO("[CommunicatorImpl][%s] opMode[%s]", __func__, currentCollOperator->opMode.Describe().c_str());
3003 2 : string needFallBackAlgName = curAlgName;
3004 2 : OpAcceleratorStateFallback();
3005 :
3006 2 : HcclResult ret = HCCL_SUCCESS;
3007 2 : switch (currentCollOperator->opMode) {
3008 1 : case OpMode::OPBASE:
3009 1 : ret = ReLoadOpbasedOp();
3010 0 : break;
3011 1 : case OpMode::OFFLOAD:
3012 1 : ret = ReLoadOffloadOp();
3013 0 : break;
3014 0 : default:
3015 0 : THROW<InternalException>(
3016 0 : StringFormat("[CommunicatorImpl::%s] OpMode error, accelerator rollback failed", __func__));
3017 : break;
3018 : }
3019 :
3020 : // 缓存当前算子的加速模式；
3021 : // 下一个算子下发时，做完算法选择后，查找上述加速模式缓存，
3022 : // 若能命中，按照上述已缓存的加速模式下发算子(大概率也是资源不足，走回退)；
3023 : // 否则，按照算法选择的加速模式下发算子。
3024 0 : opAcceStateCache.insert({{curOpParams.opType, needFallBackAlgName}, {opExecuteConfig.accState, curAlgName}});
3025 0 : HCCL_INFO("[CommunicatorImpl][%s] opAcceStateCache opType[%s], needFallBackAlgName[%s], accelerator[%s], curAlgName[%s]", __func__,
3026 : curOpParams.opType.Describe().c_str(), needFallBackAlgName.c_str(), opExecuteConfig.accState.Describe().c_str(), curAlgName.c_str());
3027 :
3028 0 : HCCL_INFO("[CommunicatorImpl][%s] end", __func__);
3029 0 : return ret;
3030 2 : }
3031 :
3032 0 : HcclResult CommunicatorImpl::GetCacheMap(AivOpCacheArgs& opCacheParam , std::shared_ptr<InsQueue>& tempInsQue)
3033 : {
3034 0 : if (hcclCacheMap_.size() > CACHEMAP_MAXSIZE) {
3035 0 : size_t clearCount = static_cast<size_t>(CACHEMAP_MAXSIZE * CACHEMAP_CLEARPERCENT);
3036 0 : for (auto it = hcclCacheMap_.begin(); clearCount > 0 && it != hcclCacheMap_.end(); clearCount--) {
3037 0 : it = hcclCacheMap_.erase(it);
3038 : }
3039 : }
3040 0 : hcclCacheMap_.emplace(std::make_pair(opCacheParam, std::move(tempInsQue)));
3041 0 : HCCL_INFO("[CommunicatorImpl][GetCacheMap]");
3042 0 : return HCCL_SUCCESS;
3043 : }
3044 :
3045 6 : HcclResult CommunicatorImpl::ReLoadOpbasedOp()
3046 : {
3047 18 : HCCL_DEBUG("[CommunicatorImpl][%s] status is [%s], isSuspended is [%d]", __func__, GetCommStatus().Describe().c_str(),
3048 : isSuspended);
3049 6 : ExecAlgSelect(curOpParams, OpMode::OPBASE); // 根据配置选择对应的collService
3050 5 : if (dynamic_cast<CollServiceDefaultImpl *>(collService) != nullptr) {
3051 3 : HCCL_ERROR("ReLoadOpbasedOp is not supported in CollServiceDefaultImpl.");
3052 1 : return HcclResult::HCCL_E_NOT_SUPPORT;
3053 : }
3054 4 : bool isAiv = (opExecuteConfig.accState == AcceleratorState::AIV || opExecuteConfig.accState == AcceleratorState::AIV_ONLY);
3055 4 : HcclResult dataTypeChkRes = OpParamsChecker::CheckOpDataTypeOpbase(curOpParams, GetOpCcuFeatureFlag(),
3056 4 : GetOpAiCpuTSFeatureFlag(), isAiv); // 算子粒度
3057 4 : if (dataTypeChkRes != HcclResult::HCCL_SUCCESS) {
3058 3 : HCCL_ERROR("[CommunicatorImpl::ReLoadOpbasedOp] DataType check fail.");
3059 1 : SetCommStatus(CommStatus::COMM_READY);
3060 1 : return dataTypeChkRes;
3061 : }
3062 :
3063 3 : if (currentCollOperator == nullptr) {
3064 3 : HCCL_ERROR("CurrentCollOperator not initialized.");
3065 1 : return HcclResult::HCCL_E_PTR;
3066 : }
3067 2 : collService->ReLoadWithOpBasedMode(*currentCollOperator);
3068 1 : return HcclResult::HCCL_SUCCESS;
3069 : }
3070 :
3071 6 : HcclResult CommunicatorImpl::ReLoadOffloadOp()
3072 : {
3073 18 : HCCL_DEBUG("[CommunicatorImpl][%s] status is [%s], isSuspended is [%d]", __func__, GetCommStatus().Describe().c_str(),
3074 : isSuspended);
3075 :
3076 6 : ExecAlgSelect(curOpParams, OpMode::OFFLOAD); // 根据配置选择对应的collService
3077 :
3078 5 : if (opExecuteConfig.accState == AcceleratorState::HOSTCPU_TS) { // 950不支持HOST_TS模式
3079 3 : HCCL_ERROR("[CommunicatorImpl::ReLoadOffloadOp] HOSTCPU_TS is not support.");
3080 1 : return HcclResult::HCCL_E_NOT_SUPPORT;
3081 : }
3082 4 : bool isAiv = (opExecuteConfig.accState == AcceleratorState::AIV || opExecuteConfig.accState == AcceleratorState::AIV_ONLY);
3083 4 : HcclResult dataTypeChkRes = OpParamsChecker::CheckOpDataTypeOffload(curOpParams, GetOpCcuFeatureFlag(),
3084 4 : GetOpAiCpuTSFeatureFlag(), isAiv); // 算子粒度
3085 4 : if (dataTypeChkRes != HcclResult::HCCL_SUCCESS) {
3086 3 : HCCL_ERROR("[CommunicatorImpl::ReLoadOffloadCollOp] DataType check fail.");
3087 1 : SetCommStatus(CommStatus::COMM_READY);
3088 1 : return dataTypeChkRes;
3089 : }
3090 :
3091 3 : if (currentCollOperator == nullptr) {
3092 3 : HCCL_ERROR("CurrentCollOperator not initialized.");
3093 1 : return HcclResult::HCCL_E_PTR;
3094 : }
3095 2 : collService->ReLoadWithOffloadMode(*currentCollOperator);
3096 1 : return HcclResult::HCCL_SUCCESS;
3097 : }
3098 :
3099 : template<typename BufferType>
3100 0 : std::shared_ptr<BufferType> CommunicatorImpl::BarrierAllocBuffer(std::size_t size)
3101 : {
3102 0 : return std::make_shared<BufferType>(size);
3103 : }
3104 :
3105 0 : HcclResult CommunicatorImpl::CreateBarrierMemory(void *&sendBuf, void *&recvBuf, uint64_t count)
3106 : {
3107 0 : HCCL_INFO("CreateBarrierMemory start.");
3108 0 : if (isFirstBarrier) {
3109 0 : barrierInMemory = BarrierAllocBuffer<DevBuffer>(count * sizeof(float));
3110 0 : barrierOutMemory = BarrierAllocBuffer<DevBuffer>(count * sizeof(float));
3111 : // 申请host侧内存，并将初始值设置为0
3112 0 : std::shared_ptr<HostBuffer> barrierHostMem = BarrierAllocBuffer<HostBuffer>(count * sizeof(float));
3113 0 : s32 sRet = memset_s(reinterpret_cast<void *>(barrierHostMem->GetAddr()), barrierHostMem->GetSize(), 0,
3114 : count * sizeof(float));
3115 0 : if (sRet != EOK) {
3116 0 : barrierInMemory.reset();
3117 0 : barrierOutMemory.reset();
3118 0 : barrierHostMem.reset();
3119 0 : HCCL_ERROR("[CreateBarrierMemory] mem set failed.errorno[%d]", sRet);
3120 0 : return HCCL_E_MEMORY;
3121 : }
3122 : // H2D拷贝
3123 0 : HrtMemcpy(reinterpret_cast<void *>(barrierInMemory->GetAddr()), barrierInMemory->GetSize(), reinterpret_cast<void *>(barrierHostMem->GetAddr()),
3124 0 : barrierHostMem->GetSize(), RT_MEMCPY_HOST_TO_DEVICE);
3125 0 : HrtMemcpy(reinterpret_cast<void *>(barrierOutMemory->GetAddr()), barrierOutMemory->GetSize(), reinterpret_cast<void *>(barrierHostMem->GetAddr()),
3126 0 : barrierHostMem->GetSize(), RT_MEMCPY_HOST_TO_DEVICE);
3127 0 : isFirstBarrier = false;
3128 0 : }
3129 : // 将内存指针赋值给传入参数
3130 0 : sendBuf = reinterpret_cast<void *>(barrierInMemory->GetAddr());
3131 0 : if (sendBuf == nullptr) {
3132 0 : HCCL_ERROR("[CreateBarrierMemory] Failed to get barrierInMemory.");
3133 0 : return HCCL_E_PTR;
3134 : }
3135 0 : recvBuf = reinterpret_cast<void *>(barrierOutMemory->GetAddr());
3136 0 : if (recvBuf == nullptr) {
3137 0 : HCCL_ERROR("[CreateBarrierMemory] Failed to get barrierOutMemory.");
3138 0 : return HCCL_E_PTR;
3139 : }
3140 0 : HCCL_INFO("CreateBarrierMemory success.");
3141 0 : return HCCL_SUCCESS;
3142 : }
3143 :
3144 0 : HcclResult CommunicatorImpl::CreateWorkspaceBuf(const char *memTag, uint64_t *size, bool *newCreated)
3145 : {
3146 0 : std::string tag = memTag != nullptr ? std::string(memTag) : "";
3147 : // empty tag is global workspace
3148 0 : if (tagWorkspaceMap_.find(tag) == tagWorkspaceMap_.end()) {
3149 0 : shared_ptr<DevBuffer> workspace = std::make_shared<DevBuffer>(*size);
3150 0 : tagWorkspaceMap_.insert(make_pair(tag, workspace));
3151 0 : HCCL_INFO("Create tagMem[%s] WorkspaceBuf success, WorkspaceBuf = %p", tag.c_str(), workspace.get());
3152 0 : if (newCreated != nullptr) {
3153 0 : *newCreated = true;
3154 : }
3155 0 : }
3156 0 : return HcclResult::HCCL_SUCCESS;
3157 0 : }
3158 :
3159 : // dpu相关
3160 39 : bool CommunicatorImpl::IsNeedDpu()
3161 : {
3162 39 : if (rankGraph == nullptr) {
3163 0 : return false;
3164 : }
3165 39 : if (rankGraph->GetPeer(myRank) == nullptr) {
3166 0 : HCCL_ERROR("[GetHostIpFromRankGraph] rankGraph peer is null!");
3167 0 : return false;
3168 : }
3169 : // 根据rankgraph直接找peer对应的ConnInterface列表
3170 39 : std::vector<std::shared_ptr<NetInstance::ConnInterface>> interfaces = rankGraph->GetPeer(myRank)->GetIfaces();
3171 54 : for (auto interface : interfaces) {
3172 20 : if (interface->GetPos() == AddrPosition::HOST) {
3173 15 : HCCL_INFO("[CommunicatorImpl][IsNeedDpu] need host dpu");
3174 5 : return true;
3175 : }
3176 20 : }
3177 34 : return false;
3178 39 : }
3179 :
3180 5 : void CommunicatorImpl::InitHccpPeer() const
3181 : {
3182 5 : RaSocketSetWhiteListStatus(1); // PEER模式需要手动开启白名单模式
3183 5 : HccpPeerManager::GetInstance().Init(devLogicId);
3184 5 : }
3185 :
3186 0 : HcclResult CommunicatorImpl::PrepareDpuKernelResource(aclrtFuncHandle &funcHandle)
3187 : {
3188 : // 获取二进制文件路径
3189 0 : std::string jsonPath;
3190 0 : std::string getPath = getenv("ASCEND_HOME_PATH");
3191 0 : if (!getPath.empty()) {
3192 0 : jsonPath = getPath;
3193 : } else {
3194 0 : jsonPath = "/usr/local/Ascend/cann/";
3195 0 : HCCL_WARNING("[CommunicatorImpl::%s] ENV:ASCEND_HOME_PATH is not set", __func__);
3196 : }
3197 :
3198 0 : jsonPath += "/opp/built-in/op_impl/dpu/";
3199 0 : HCCL_DEBUG("[CommunicatorImpl::%s] kernel folder path[%s]", __func__, jsonPath.c_str());
3200 :
3201 : // cpuKernelMode为1时，json命名需与so命名保持一致， 即libccl_dpu.json与libccl_dpu.so
3202 0 : jsonPath += "libccl_dpu.json";
3203 0 : char realPath[PATH_MAX] = {0};
3204 0 : CHK_PRT_RET(realpath(jsonPath.c_str(), realPath) == nullptr,
3205 : HCCL_ERROR("[CommunicatorImpl::%s]: %s is not a valid real path, err[%d]", __func__, jsonPath.c_str(), errno),
3206 : HCCL_E_INTERNAL);
3207 0 : HCCL_INFO("[CommunicatorImpl::%s] realPath: %s", __func__, realPath);
3208 :
3209 : aclrtBinHandle binHandle;
3210 : aclrtBinaryLoadOptions options;
3211 : aclrtBinaryLoadOption option;
3212 0 : option.type = ACL_RT_BINARY_LOAD_OPT_CPU_KERNEL_MODE; // AI CPU算子注册模式 ????
3213 0 : option.value.cpuKernelMode = 1; // 0 ：仅需要加载json，1 ：加载cpu so & json，2: LoadFromData
3214 0 : options.numOpt = 1;
3215 0 : options.options = &option;
3216 0 : if (aclrtBinaryLoadFromFile(realPath, &options, &binHandle) != ACL_SUCCESS) {
3217 0 : HCCL_ERROR("[CommunicatorImpl::%s] load binary from file error.", __func__);
3218 0 : return HCCL_E_OPEN_FILE_FAILURE;
3219 : }
3220 :
3221 : // 创建dpustream
3222 0 : if (aclrtCreateStreamWithConfig(&dpuStream, 0, ACL_STREAM_FAST_LAUNCH) != ACL_SUCCESS) {
3223 0 : HCCL_ERROR("[CommunicatorImpl::%s] Create Local Stream Failed", __func__);
3224 0 : return HCCL_E_INTERNAL;
3225 : }
3226 :
3227 : // 查找核函数
3228 0 : if (aclrtBinaryGetFunction(binHandle, "RunDpuRpcSrvLaunch", &funcHandle) != ACL_SUCCESS) {
3229 0 : HCCL_ERROR("[CommunicatorImpl::%s] Get Function Failed", __func__);
3230 0 : return HCCL_E_INTERNAL;
3231 : }
3232 :
3233 0 : return HCCL_SUCCESS;
3234 0 : }
3235 :
3236 2 : HcclResult CommunicatorImpl::LaunchDpuKernel(aclrtFuncHandle &funcHandle)
3237 : {
3238 : // 下发
3239 6 : HCCL_INFO("[CommunicatorImpl::%s] Launch Dpu Kernel", __func__);
3240 : aclrtLaunchKernelCfg cfg;
3241 : aclrtLaunchKernelAttr kernelAttr;
3242 2 : kernelAttr.id = ACL_RT_LAUNCH_KERNEL_ATTR_TIMEOUT;
3243 2 : kernelAttr.value.timeout = NOTIFY_DEFAULT_WAIT_TIME > std::numeric_limits<uint16_t>::max() ?
3244 0 : std::numeric_limits<uint16_t>::max() : NOTIFY_DEFAULT_WAIT_TIME;
3245 2 : cfg.numAttrs = 1;
3246 2 : cfg.attrs = &kernelAttr;
3247 2 : constexpr u32 numBlocks = 1;
3248 2 : hostArgs.commId = id;
3249 2 : hostArgs.memorySize = SHARE_HBM_MEMORY_SIZE;
3250 2 : hostArgs.hostMem = hostShareBuf;
3251 8 : hostArgs.shareHBM = tagDpuShmemArgsMap_[DPUTAG].connectType_ == HOST_DEVICE_CONNECT_TYPE_UB ? tagDpuShmemArgsMap_[DPUTAG].va_ : tagDpuShmemArgsMap_[DPUTAG].accessVA_;
3252 2 : hostArgs.deviceId = devLogicId;
3253 8 : hostArgs.taskexceptionVa = tagDpuShmemArgsMap_[DPUTASKEXCEPTION].connectType_ == HOST_DEVICE_CONNECT_TYPE_UB ? tagDpuShmemArgsMap_[DPUTASKEXCEPTION].va_ : tagDpuShmemArgsMap_[DPUTASKEXCEPTION].accessVA_;
3254 6 : HCCL_INFO("[CommunicatorImpl::%s] DpuKernelLaunchParam{commId:%s; memorySize:%u; shareHBM:%p; taskexceptionVa:%p}",
3255 : __func__, hostArgs.commId.c_str(), hostArgs.memorySize, hostArgs.shareHBM, hostArgs.taskexceptionVa);
3256 2 : CHK_RET(SaveDpuStreamId());
3257 2 : size_t argsSize = sizeof(hostArgs);
3258 : aclrtPlaceHolderInfo placeHolderArrays;
3259 2 : size_t placeHolderNum = 0;
3260 2 : if (aclrtLaunchKernelWithHostArgs(funcHandle, numBlocks, dpuStream, &cfg, &hostArgs, argsSize,
3261 : &placeHolderArrays, placeHolderNum)
3262 2 : != ACL_SUCCESS) {
3263 0 : HCCL_ERROR("[CommunicatorImpl::%s] Launch Dpu Kernel Failed", __func__);
3264 0 : return HCCL_E_INTERNAL;
3265 : }
3266 2 : return HCCL_SUCCESS;
3267 : }
3268 :
3269 3 : HcclResult CommunicatorImpl::InitAndLaunchDpuKernel()
3270 : {
3271 9 : HCCL_INFO("[CommunicatorImpl::%s] Start to Launch Dpu Kernel", __func__);
3272 : // 申请共享内存(需要在npu ctx 下进行)
3273 9 : for (auto &tmpShmem : tagDpuShmemArgsMap_) {
3274 6 : bool newCreate = false;
3275 6 : uint64_t memSize = tmpShmem.first == DPUTAG ? static_cast<uint64_t>(SHARE_HBM_MEMORY_SIZE) : static_cast<uint64_t>(DPU_TASKEXCEPTION_MEMORY_SIZE);
3276 6 : HcclResult memRet = GetKFCWorkSpaceVA(tmpShmem.first, &memSize, &tmpShmem.second.accessVA_, &newCreate);
3277 6 : if (memRet != HCCL_SUCCESS) {
3278 0 : HCCL_ERROR("[CommunicatorImpl::InitCommResource] Alloc Share HBM Failed");
3279 0 : return HCCL_E_RUNTIME;
3280 : }
3281 : }
3282 : // 设置XPU
3283 9 : HCCL_INFO("[CommunicatorImpl::%s] Switch to Dpu Ctx", __func__);
3284 3 : if (aclrtGetCurrentContext(&npuContext) != ACL_SUCCESS) {
3285 0 : HCCL_ERROR("[CommunicatorImpl::%s] Get Npu Ctx Failed", __func__);
3286 0 : return HCCL_E_INTERNAL;
3287 : }
3288 3 : if (HrtSetXpuDevice(TEMP_DEV_TYPE_DPU, 0) != HCCL_SUCCESS) {
3289 0 : HCCL_ERROR("[CommunicatorImpl::%s] Switch to Dpu Ctx Failed", __func__);
3290 0 : return HCCL_E_INTERNAL;
3291 : }
3292 3 : if (aclrtGetCurrentContext(&dpuContext) != ACL_SUCCESS) {
3293 0 : HCCL_ERROR("[CommunicatorImpl::%s] Get Dpu Ctx Failed", __func__);
3294 0 : return HCCL_E_INTERNAL;
3295 : }
3296 :
3297 : // 准备资源
3298 : aclrtFuncHandle funcHandle;
3299 3 : CHK_RET(PrepareDpuKernelResource(funcHandle));
3300 :
3301 3 : hostShareBuf = malloc(SHARE_HBM_MEMORY_SIZE);
3302 3 : CHK_PTR_NULL(hostShareBuf);
3303 :
3304 : // 下发
3305 3 : HcclResult ret = LaunchDpuKernel(funcHandle);
3306 3 : if (ret != HCCL_SUCCESS) {
3307 3 : HCCL_ERROR("[CommunicatorImpl::%s] Launch Dpu Kernel Failed", __func__);
3308 1 : free(hostShareBuf);
3309 1 : hostShareBuf = nullptr;
3310 1 : return ret;
3311 : }
3312 :
3313 : // 切换回当前Ctx
3314 6 : HCCL_INFO("[CommunicatorImpl::%s] Switch to Npu Ctx", __func__);
3315 2 : if (ACL_SUCCESS != aclrtSetCurrentContext(npuContext)) {
3316 0 : HCCL_ERROR("[CommunicatorImpl::%s] Reset Current Ctx Failed", __func__);
3317 0 : free(hostShareBuf);
3318 0 : hostShareBuf = nullptr;
3319 0 : return HCCL_E_INTERNAL;
3320 : }
3321 :
3322 6 : HCCL_INFO("[CommunicatorImpl::%s] Launch Dpu Kernel End", __func__);
3323 2 : isDpuKernelLaunched = true;
3324 2 : g_commNum++;
3325 2 : return HCCL_SUCCESS;
3326 : }
3327 :
3328 4 : HcclResult CommunicatorImpl::InitAndLaunchAicpuKernel()
3329 : {
3330 12 : HCCL_INFO("[CommunicatorImpl::%s] Start to Launch Aicpu Kernel", __func__);
3331 : // context切换
3332 :
3333 : // 准备资源
3334 4 : std::string kernelName = "HcclDpuTaskexpShmemRestore";
3335 4 : aclrtFuncHandle funcHandle = GetAicpuKernelFuncHandle(kernelName.c_str());
3336 4 : constexpr u32 numBlocks = 1;
3337 : aclrtStream tempAicpuStream; // 创建局部流
3338 4 : if (aclrtCreateStreamWithConfig(&tempAicpuStream, 0, ACL_STREAM_FAST_LAUNCH) != ACL_SUCCESS) { // 后两个入参？
3339 3 : HCCL_ERROR("[CommunicatorImpl::%s] Create Local Stream Failed", __func__);
3340 1 : return HCCL_E_INTERNAL;
3341 : }
3342 : aclrtLaunchKernelCfg cfg;
3343 : aclrtLaunchKernelAttr kernelAttr;
3344 3 : kernelAttr.id = ACL_RT_LAUNCH_KERNEL_ATTR_TIMEOUT;
3345 3 : kernelAttr.value.timeout = NOTIFY_DEFAULT_WAIT_TIME > std::numeric_limits<uint16_t>::max() ?
3346 0 : std::numeric_limits<uint16_t>::max() : NOTIFY_DEFAULT_WAIT_TIME;
3347 3 : cfg.numAttrs = 1;
3348 3 : cfg.attrs = &kernelAttr;
3349 : struct AicpuKernelLaunchParam {
3350 : char commId[COMM_NAME_MAX_LENGTH];
3351 : void *taskexceptionVa;
3352 : u64 memorySize;
3353 : uint32_t deviceId;
3354 : };
3355 : AicpuKernelLaunchParam hostArgsTmp;
3356 3 : auto sRet = strncpy_s(hostArgsTmp.commId, COMM_NAME_MAX_LENGTH, id.c_str(), COMM_NAME_MAX_LENGTH - 1);
3357 3 : if (sRet != 0) {
3358 0 : HCCL_ERROR("[CommunicatorImpl::InitAndLaunchAicpuKernel] commId strncpy_s failed, ret[%d]", sRet);
3359 0 : return HCCL_E_INTERNAL;
3360 : }
3361 6 : hostArgsTmp.taskexceptionVa = tagDpuShmemArgsMap_[DPUTASKEXCEPTION].accessVA_;
3362 3 : hostArgsTmp.memorySize = DPU_TASKEXCEPTION_MEMORY_SIZE;
3363 3 : hostArgsTmp.deviceId = devLogicId;
3364 3 : size_t argsSize = sizeof(hostArgsTmp);
3365 : aclrtPlaceHolderInfo placeHolderArrays;
3366 3 : size_t placeHolderNum = 0;
3367 : // 下发
3368 3 : rtError_t ret = aclrtLaunchKernelWithHostArgs(funcHandle, numBlocks, tempAicpuStream, &cfg, &hostArgsTmp, argsSize,
3369 : &placeHolderArrays, placeHolderNum);
3370 3 : if (ret != RT_ERROR_NONE) {
3371 3 : HCCL_ERROR("[CommunicatorImpl::%s] Launch Aicpu Kernel Failed", __func__);
3372 1 : return HCCL_E_INTERNAL;
3373 : }
3374 2 : HcclStreamSynchronize(tempAicpuStream); // 同步
3375 2 : aclError aclRet = aclrtDestroyStreamForce(tempAicpuStream); // 销毁局部流
3376 2 : if (ACL_SUCCESS != aclRet) {
3377 3 : HCCL_ERROR("Destroy Stream Failed, aclReturn[%d]", aclRet);
3378 1 : return HCCL_E_RUNTIME;
3379 : }
3380 3 : HCCL_INFO("[CommunicatorImpl::%s] Launch Aicpu Kernel End", __func__);
3381 1 : return HCCL_SUCCESS;
3382 4 : }
3383 :
3384 40 : void CommunicatorImpl::AppendLocalDieIdForLinks()
3385 : {
3386 40 : if (rankSize == 1) {
3387 3 : HCCL_INFO("[AppendLocalDieIdForLinks] rankSize = 1, No RankGraph exists");
3388 1 : return;
3389 : }
3390 :
3391 39 : auto srcRankNode = rankGraph->GetPeer(myRank)->GetNodeId();
3392 :
3393 140 : auto processLinks = [&](const std::vector<std::shared_ptr<NetInstance::Link>>& links, bool isSource) {
3394 190 : for (auto link : links) {
3395 50 : auto iface = isSource ? link->GetSourceIface() : link->GetTargetIface();
3396 50 : if (iface->GetPos() == AddrPosition::HOST || *(iface->GetLinkProtocols().begin()) == LinkProtocol::PCIE) {
3397 10 : continue;
3398 : }
3399 40 : u32 dieId = GetLocalDieId({myRank, *iface}, *(link->GetLinkProtocols().begin()));
3400 120 : HCCL_INFO("[CommunicatorImpl][AppendLocalDieIdForLinks] get link dieid[%u]", dieId);
3401 40 : iface->SetLocalDieId(dieId);
3402 60 : }
3403 140 : };
3404 :
3405 54 : for (auto level : rankGraph->GetLevels(myRank)) {
3406 15 : auto netInstance = rankGraph->GetNetInstanceByRankId(level, myRank);
3407 15 : auto& vGraph = netInstance->GetGraph();
3408 :
3409 : // Process fabric links
3410 25 : for (auto fabric : netInstance->GetFabrics()) {
3411 10 : auto dstRankNode = fabric->GetNodeId();
3412 10 : processLinks(vGraph.GetEdges(srcRankNode, dstRankNode), true);
3413 10 : processLinks(vGraph.GetEdges(dstRankNode, srcRankNode), false);
3414 10 : }
3415 :
3416 : // Process direct peer links
3417 75 : for (u32 dstRank = 0; dstRank < rankSize; ++dstRank) {
3418 60 : auto dstRankNode = rankGraph->GetPeer(dstRank)->GetNodeId();
3419 60 : processLinks(vGraph.GetEdges(srcRankNode, dstRankNode), true);
3420 60 : processLinks(vGraph.GetEdges(dstRankNode, srcRankNode), false);
3421 : }
3422 39 : }
3423 : }
3424 :
3425 1 : HcclResult CommunicatorImpl::GetLocalCclBuffer(void **addr, uint64_t *size)
3426 : {
3427 1 : CHK_PTR_NULL(inCclBuffer.get());
3428 1 : *addr = reinterpret_cast<void*>(inCclBuffer.get()->GetAddr());
3429 1 : *size = static_cast<uint64_t>(inCclBuffer.get()->GetSize());
3430 3 : HCCL_INFO("CommunicatorImpl::GetLocalCclBuffer success, addr[%p], size[%llu]", *addr, *size);
3431 1 : return HcclResult::HCCL_SUCCESS;
3432 : }
3433 :
3434 5 : HcclResult CommunicatorImpl::GetDevMemWorkSpace(const std::string &memTag, uint64_t *size, void **addr, bool *newCreated)
3435 : {
3436 5 : if (memTag == DPUTAG) {
3437 0 : return GetKFCWorkSpaceVA(memTag, size, addr, newCreated);
3438 : }
3439 5 : auto iter = tagWorkspaceMap_.find(memTag);
3440 5 : if (iter != tagWorkspaceMap_.end()) {
3441 2 : std::shared_ptr<DevBuffer> oldWorkspace = iter->second;
3442 2 : if (*size != static_cast<uint64_t>(oldWorkspace.get()->GetSize())) {
3443 3 : HCCL_ERROR("HcclCommunicator::GetDevMemWorkSpace, The size of oldWorkspace %p is non-consistent, target size compare now size: %llu->%llu", *addr, *size, oldWorkspace.get()->GetSize());
3444 1 : return HCCL_E_PARA;
3445 : }
3446 1 : *addr = reinterpret_cast<void *>(oldWorkspace.get()->GetAddr());
3447 1 : if (newCreated != nullptr) {
3448 1 : *newCreated = false;
3449 : }
3450 1 : return HcclResult::HCCL_SUCCESS;
3451 2 : }
3452 :
3453 3 : shared_ptr<DevBuffer> newWorkspace = std::make_shared<DevBuffer>(*size);
3454 3 : tagWorkspaceMap_.insert(make_pair(memTag, newWorkspace));
3455 9 : HCCL_INFO("Create tagMem[%s] WorkspaceBuf success, WorkspaceBuf: %p -> %p, size[%llu]", memTag.c_str(), newWorkspace.get(), newWorkspace.get()->GetAddr(), *size);
3456 3 : if (newCreated != nullptr) {
3457 2 : *newCreated = true;
3458 : }
3459 3 : *addr = reinterpret_cast<void *>(newWorkspace.get()->GetAddr());
3460 3 : return HcclResult::HCCL_SUCCESS;
3461 3 : }
3462 :
3463 4 : HcclResult CommunicatorImpl::AllocAndRegKFCWorkSpace(uint64_t size, const std::string &memTag) // 完成共享内存的申请和注册
3464 : {
3465 4 : auto it = tagDpuShmemArgsMap_.find(memTag);
3466 4 : if (it == tagDpuShmemArgsMap_.end()) {
3467 0 : HCCL_ERROR("memTag is invalid, memTag: %s", memTag.c_str());
3468 0 : return HCCL_E_PARA;
3469 : }
3470 :
3471 4 : if (it->second.accessVA_ != nullptr && it->second.va_ != nullptr) {
3472 0 : HCCL_WARNING("[CommunicatorImpl::%s] memTag[%s] accessVA_ is not nullptr, maybe already register, accessVA_: %p", __func__, memTag.c_str(), it->second.accessVA_);
3473 0 : CHK_RET(DestroyKFCWorkSpaceVA());
3474 : }
3475 4 : int32_t deviceLogicId = 0;
3476 4 : aclError aclRet = aclrtGetLogicDevIdByUserDevId(devLogicId, &deviceLogicId); // userDevId 转 logicDevId
3477 4 : if (aclRet != ACL_SUCCESS) {
3478 0 : HCCL_ERROR("[CommunicatorImpl::%s] aclrtGetLogicDevIdByUserDevId failed, devLogicId: %u, ret: %d", __func__, devLogicId, aclRet);
3479 0 : return HCCL_E_RUNTIME;
3480 : }
3481 4 : CHK_RET(HrtHalGetDeviceInfo(deviceLogicId, MODULE_TYPE_SYSTEM, INFO_TYPE_HD_CONNECT_TYPE, it->second.connectType_));
3482 12 : HCCL_INFO("[CommunicatorImpl::%s] connectType_: %ld", __func__, it->second.connectType_);
3483 :
3484 4 : HcclResult ret = HCCL_SUCCESS;
3485 4 : if (it->second.connectType_ == HOST_DEVICE_CONNECT_TYPE_PCIE) {
3486 2 : it->second.va_ = HrtMalloc(size, ACL_MEM_TYPE_HIGH_BAND_WIDTH);
3487 2 : ret = HrtHalHostRegister(it->second.va_, size, DEV_SVM_MAP_HOST, deviceLogicId, &it->second.accessVA_);
3488 2 : } else if (it->second.connectType_ == HOST_DEVICE_CONNECT_TYPE_UB) {
3489 1 : it->second.originVa_ = malloc(size + ALIGN_4K); // 保证足够的对齐余量（最多偏移 ALIGN_4K-1）
3490 1 : CHK_PTR_NULL(it->second.originVa_);
3491 1 : void* ptr = it->second.originVa_;
3492 1 : std::size_t storeSize = 0; // 不预留对象空间，仅做指针向上对齐
3493 1 : std::size_t space = static_cast<std::size_t>(size + ALIGN_4K);
3494 : // 如果对齐失败，va_为nullptr，在HrtHalHostRegister中校验返回HCCL_E_PTR
3495 1 : it->second.va_ = std::align(static_cast<std::size_t>(ALIGN_4K), storeSize, ptr, space);
3496 1 : ret = HrtHalHostRegister(it->second.va_, size, HOST_MEM_MAP_DEV_PCIE_TH, deviceLogicId, &it->second.accessVA_);
3497 : } else {
3498 1 : return HCCL_E_NOT_SUPPORT;
3499 : }
3500 3 : if (ret != HCCL_SUCCESS) {
3501 1 : return CleanupKFCWorkSpaceOnFailure(it->second, ret);
3502 : }
3503 2 : auto ptr = it->second.connectType_ == HOST_DEVICE_CONNECT_TYPE_UB ? it->second.va_ : it->second.accessVA_;
3504 2 : errno_t cpyRet = memset_s(ptr, sizeof(uint8_t), 0, sizeof(uint8_t)); // 首字节作为信号标志位
3505 2 : if (cpyRet != EOK) {
3506 0 : HCCL_ERROR("[CommunicatorImpl::AllocAndRegKFCWorkSpace] set 0 failed: %d", cpyRet);
3507 0 : return HCCL_E_MEMORY; // 如果失败，会在~CommunicatorImpl中进行解注册与内存释放
3508 : }
3509 6 : HCCL_INFO("CommunicatorImpl::AllocAndRegKFCWorkSpace va_[%p], accessVA_[%p]", it->second.va_, it->second.accessVA_);
3510 2 : return HCCL_SUCCESS;
3511 : }
3512 :
3513 1 : HcclResult CommunicatorImpl::CleanupKFCWorkSpaceOnFailure(DpuShmem &shmem, HcclResult ret)
3514 : {
3515 3 : HCCL_ERROR("[CommunicatorImpl::%s] HrtHalHostRegister failed, ret: %d, connect type: %ld", __func__, ret,
3516 : shmem.connectType_);
3517 1 : if (shmem.va_ != nullptr) {
3518 1 : if (shmem.connectType_ == HOST_DEVICE_CONNECT_TYPE_PCIE) {
3519 1 : HrtFree(shmem.va_);
3520 0 : } else if (shmem.connectType_ == HOST_DEVICE_CONNECT_TYPE_UB) {
3521 0 : free(shmem.originVa_);
3522 0 : shmem.originVa_ = nullptr;
3523 : }
3524 1 : shmem.va_ = nullptr;
3525 : }
3526 1 : shmem.accessVA_ = nullptr;
3527 1 : return ret;
3528 : }
3529 :
3530 7 : HcclResult CommunicatorImpl::GetKFCWorkSpaceVA(const std::string &memTag, const uint64_t *size, void **addr, bool *newCreated)
3531 : {
3532 21 : HCCL_INFO("CommunicatorImpl::GetKFCWorkSpaceVA tag[%s]", memTag.c_str()); // 调试日志
3533 7 : auto it = tagDpuShmemArgsMap_.find(memTag);
3534 7 : if (it == tagDpuShmemArgsMap_.end()) {
3535 0 : HCCL_ERROR("memTag is invalid, memTag: %s", memTag.c_str());
3536 0 : return HCCL_E_PARA;
3537 : }
3538 7 : auto iter = tagWorkspaceVAMap_.find(memTag);
3539 7 : if (iter != tagWorkspaceVAMap_.end()) {
3540 3 : std::shared_ptr<DevBuffer> oldWorkspace = iter->second;
3541 3 : if (*size != static_cast<uint64_t>(oldWorkspace.get()->GetSize())) {
3542 3 : HCCL_ERROR("HcclCommunicator::GetKFCWorkSpaceVA, The size of oldWorkspace %p is non-consistent, target size compare now size: %llu->%llu", *addr, *size, oldWorkspace.get()->GetSize());
3543 1 : return HCCL_E_PARA;
3544 : }
3545 2 : *addr = reinterpret_cast<void *>(oldWorkspace.get()->GetAddr());
3546 2 : if (newCreated != nullptr) {
3547 1 : *newCreated = false;
3548 : }
3549 2 : return HcclResult::HCCL_SUCCESS;
3550 3 : }
3551 :
3552 10 : CHK_RET(AllocAndRegKFCWorkSpace(*size, memTag));
3553 2 : shared_ptr<DevBuffer> newWorkspace = DevBuffer::Create(reinterpret_cast<uintptr_t>(it->second.accessVA_), *size);
3554 2 : tagWorkspaceVAMap_.insert(make_pair(memTag, newWorkspace));
3555 2 : if (newCreated != nullptr) {
3556 2 : *newCreated = true;
3557 : }
3558 2 : *addr = reinterpret_cast<void *>(newWorkspace.get()->GetAddr());
3559 2 : return HcclResult::HCCL_SUCCESS;
3560 2 : }
3561 :
3562 844 : HcclResult CommunicatorImpl::DestroyKFCWorkSpaceVA()
3563 : {
3564 2532 : for (auto &tmpShmem : tagDpuShmemArgsMap_) {
3565 1688 : if (tmpShmem.second.accessVA_ == nullptr && tmpShmem.second.va_ == nullptr) {
3566 1677 : continue;
3567 : }
3568 11 : int32_t deviceLogicId = 0;
3569 11 : aclError aclRet = aclrtGetLogicDevIdByUserDevId(devLogicId, &deviceLogicId); // userDevId 转 logicDevId
3570 11 : if (aclRet != ACL_SUCCESS) {
3571 0 : HCCL_ERROR("[CommunicatorImpl::%s] aclrtGetLogicDevIdByUserDevId failed, devLogicId: %u, ret: %d", __func__, devLogicId, aclRet);
3572 0 : return HCCL_E_RUNTIME;
3573 : }
3574 :
3575 : // 必须先halHostUnregister解除映射，再释放设备内存，否则HrtFree会因内存被pin住而异常
3576 11 : if (tmpShmem.second.va_ != nullptr) {
3577 6 : HcclResult ret = HrtHalHostUnregister(tmpShmem.second.va_, deviceLogicId);
3578 6 : if (ret != HCCL_SUCCESS) {
3579 0 : HCCL_ERROR("[CommunicatorImpl::%s] HrtHalHostUnregister failed, ret[%d]", __func__, ret);
3580 : }
3581 :
3582 6 : if (tmpShmem.second.connectType_ == HOST_DEVICE_CONNECT_TYPE_PCIE) {
3583 4 : DECTOR_TRY_CATCH("CommunicatorImpl", HrtFree(tmpShmem.second.va_));
3584 2 : } else if (tmpShmem.second.connectType_ == HOST_DEVICE_CONNECT_TYPE_UB) {
3585 2 : DECTOR_TRY_CATCH("CommunicatorImpl", free(tmpShmem.second.originVa_));
3586 2 : tmpShmem.second.originVa_ = nullptr;
3587 : }
3588 : }
3589 :
3590 11 : tmpShmem.second.va_ = nullptr;
3591 11 : tmpShmem.second.accessVA_ = nullptr;
3592 11 : tagWorkspaceVAMap_.erase(tmpShmem.first);
3593 : }
3594 844 : return HCCL_SUCCESS;
3595 : }
3596 :
3597 0 : HcclResult CommunicatorImpl::GetAicpuOpStreamNotify(rtStream_t *opStream, u8 aicpuNotifyNum, void** aicpuNotify) const
3598 : {
3599 0 : GetAicpuStreamManager().AllocFreeStream();
3600 0 : Stream *stream = GetAicpuStreamManager().GetFreeStream();
3601 0 : *opStream = stream->GetPtr();
3602 0 : GetHostDeviceSyncNotifyManager().GetMc2AiCpuNotifys(aicpuNotifyNum, aicpuNotify);
3603 0 : return HCCL_SUCCESS;
3604 : }
3605 :
3606 1 : HcclResult CommunicatorImpl::GetNetLayers(uint32_t **netLayers, uint32_t *netLayerNum)
3607 : {
3608 : try {
3609 1 : CHK_PTR_NULL(rankGraph);
3610 1 : u32 rankId = rankGraph->GetMyRank();
3611 1 : std::set<u32> levels = rankGraph->GetLevels(rankId);
3612 1 : u32 num = rankGraph->GetLevelNum();
3613 1 : netLayersVec.clear();
3614 1 : netLayersVec = std::vector<u32>(levels.begin(), levels.end());
3615 1 : *netLayers = netLayersVec.data();
3616 1 : *netLayerNum = num;
3617 1 : return HCCL_SUCCESS;
3618 1 : } catch (const InvalidParamsException& e) {
3619 0 : HCCL_ERROR(e.what());
3620 0 : return HCCL_E_PARA;
3621 0 : } catch (const NullPtrException& e) {
3622 0 : HCCL_ERROR(e.what());
3623 0 : return HCCL_E_PTR;
3624 0 : } catch (const std::exception& e) {
3625 0 : HCCL_ERROR(e.what());
3626 0 : return HCCL_E_INTERNAL;
3627 0 : }
3628 : }
3629 :
3630 2 : HcclResult CommunicatorImpl::GetInstRanksByNetLayer(uint32_t netLayer, uint32_t **ranks, uint32_t *rankNum)
3631 : {
3632 2 : CHK_PTR_NULL(rankGraph);
3633 2 : u32 num = 0;
3634 2 : rankListVec.clear();
3635 65 : TRY_CATCH_RETURN(rankGraph->GetLocalInstRanks(netLayer, rankListVec, num));
3636 1 : *ranks = rankListVec.data();
3637 1 : *rankNum = num;
3638 1 : return HCCL_SUCCESS;
3639 : }
3640 :
3641 7 : HcclResult CommunicatorImpl::GetInstTopoTypeByNetLayer(uint32_t netLayer, uint32_t* topoType)
3642 : {
3643 7 : CHK_PTR_NULL(rankGraph);
3644 70 : TRY_CATCH_RETURN(rankGraph->GetNetType(netLayer));
3645 6 : auto type = rankGraph->GetNetType(netLayer);
3646 : static const std::unordered_map<NetType, uint32_t> netTypeMap = {
3647 : {NetType::CLOS, static_cast<uint32_t>(CommTopo::COMM_TOPO_CLOS)},
3648 : {NetType::MESH_1D, static_cast<uint32_t>(CommTopo::COMM_TOPO_1DMESH)},
3649 : {NetType::A3_SERVER, static_cast<uint32_t>(CommTopo::COMM_TOPO_910_93)},
3650 : {NetType::A2_AX_SERVER, static_cast<uint32_t>(CommTopo::COMM_TOPO_A2AXSERVER)},
3651 8 : {NetType::TOPO_FILE_DESC, static_cast<uint32_t>(CommTopo::COMM_TOPO_CUSTOM)}};
3652 :
3653 6 : auto it = netTypeMap.find(type);
3654 6 : if (it != netTypeMap.end()) {
3655 6 : *topoType = it->second;
3656 6 : return HCCL_SUCCESS;
3657 : }
3658 0 : return HCCL_E_PARA;
3659 : }
3660 :
3661 4 : HcclResult CommunicatorImpl::GetInstSizeListByNetLayer(uint32_t netLayer, uint32_t** instSizeList, uint32_t* listSize)
3662 : {
3663 : try {
3664 10 : CHK_PTR_NULL(rankGraph);
3665 2 : u32 size = 0;
3666 2 : instSizeVec.clear();
3667 2 : auto ret = rankGraph->GetNetInstanceList(netLayer, instSizeVec, size);
3668 2 : if (ret != HCCL_SUCCESS) {
3669 3 : HCCL_ERROR("[CommunicatorImpl::GetInstSizeListByNetLayer] Failed to get instSzie[%p] at netLayer[%u]",
3670 : listSize, netLayer);
3671 1 : return ret;
3672 : }
3673 1 : *instSizeList = instSizeVec.data();
3674 1 : *listSize = size;
3675 1 : return HCCL_SUCCESS;
3676 0 : } catch (const InvalidParamsException& e) {
3677 0 : HCCL_ERROR(e.what());
3678 0 : return HCCL_E_PARA;
3679 0 : } catch (const NullPtrException& e) {
3680 0 : HCCL_ERROR(e.what());
3681 0 : return HCCL_E_PTR;
3682 0 : } catch (const std::exception& e) {
3683 0 : HCCL_ERROR(e.what());
3684 0 : return HCCL_E_INTERNAL;
3685 0 : }
3686 : }
3687 :
3688 :
3689 3 : static HcclResult InsertInnerLink(const NetInstance::Path& path, std::vector<CommLink>& linkListVec)
3690 : {
3691 6 : for (const auto& link : path.links) {
3692 3 : const NetInstance::Link *peer2peer = &link;
3693 6 : for (LinkProtocol protocol : link.GetLinkProtocols()) {
3694 : CommLink commLink;
3695 3 : CommLinkInit(&commLink, 1);
3696 3 : const CommProtocol &commProtocol = LinkProtocolToCommProtocol(protocol);
3697 3 : commLink.linkAttr.linkProtocol = commProtocol;
3698 3 : commLink.linkAttr.hop = peer2peer->GetHop();
3699 3 : commLink.srcEndpointDesc.protocol = commProtocol;
3700 3 : commLink.dstEndpointDesc.protocol = commProtocol;
3701 :
3702 : // 设置源端点
3703 3 : std::shared_ptr<NetInstance::ConnInterface> srcConnInterface = link.GetSourceIface();
3704 3 : CHK_PTR_NULL(srcConnInterface);
3705 3 : HcclResult result = GetCommAddr(commLink.srcEndpointDesc.commAddr, srcConnInterface->GetAddr());
3706 3 : if (result != HCCL_SUCCESS)
3707 0 : return result;
3708 :
3709 : // 设置目标端点
3710 3 : std::shared_ptr<NetInstance::ConnInterface> dstConnInterface = link.GetTargetIface();
3711 3 : CHK_PTR_NULL(dstConnInterface);
3712 3 : result = GetCommAddr(commLink.dstEndpointDesc.commAddr, dstConnInterface->GetAddr());
3713 3 : if (result != HCCL_SUCCESS)
3714 0 : return result;
3715 :
3716 3 : linkListVec.emplace_back(std::move(commLink));
3717 6 : }
3718 : }
3719 :
3720 3 : return HCCL_SUCCESS;
3721 : }
3722 :
3723 12 : static HcclResult InsertClosLinks(const NetInstance::Path &path, std::vector<CommLink> &linkListVec)
3724 : {
3725 12 : const NetInstance::Link *peer2net = nullptr;
3726 12 : const NetInstance::Link *net2peer = nullptr;
3727 36 : for (const auto &link : path.links) {
3728 24 : bool srcNull = (link.GetSourceIface() == nullptr);
3729 24 : bool dstNull = (link.GetTargetIface() == nullptr);
3730 24 : if (!srcNull && dstNull) {
3731 12 : peer2net = &link ;
3732 12 : } else if (srcNull && !dstNull) {
3733 12 : net2peer = &link ;
3734 : }
3735 : }
3736 12 : auto srcInterface = peer2net->GetSourceIface();
3737 12 : auto dstInterface = net2peer->GetTargetIface();
3738 12 : CHK_PTR_NULL(srcInterface);
3739 12 : CHK_PTR_NULL(dstInterface);
3740 24 : for (LinkProtocol protocol : peer2net->GetLinkProtocols()) {
3741 : CommLink commLink;
3742 12 : CommLinkInit(&commLink, 1);
3743 12 : const CommProtocol &commProtocol = LinkProtocolToCommProtocol(protocol);
3744 :
3745 12 : commLink.linkAttr.linkProtocol = commProtocol;
3746 12 : commLink.linkAttr.hop = peer2net->GetHop();
3747 :
3748 12 : commLink.srcEndpointDesc.protocol = commProtocol;
3749 12 : commLink.dstEndpointDesc.protocol = commProtocol;
3750 :
3751 : // 设置源端点
3752 12 : HcclResult result = GetCommAddr(commLink.srcEndpointDesc.commAddr, srcInterface->GetAddr());
3753 12 : if (result != HCCL_SUCCESS)
3754 0 : return result;
3755 : // 设置目标端点
3756 12 : result = GetCommAddr(commLink.dstEndpointDesc.commAddr, dstInterface->GetAddr());
3757 12 : if (result != HCCL_SUCCESS)
3758 0 : return result;
3759 12 : linkListVec.emplace_back(std::move(commLink));
3760 12 : }
3761 12 : return HCCL_SUCCESS;
3762 12 : }
3763 :
3764 5 : HcclResult CommunicatorImpl::GetLinks(uint32_t netLayer, uint32_t srcRank, uint32_t dstRank, CommLink** linkList,
3765 : uint32_t* listSize)
3766 : {
3767 : try {
3768 5 : CHK_PTR_NULL(rankGraph);
3769 5 : std::vector<NetInstance::Path> paths = rankGraph->GetPaths(netLayer, srcRank, dstRank);
3770 5 : linkListVec.clear();
3771 : // 遍历每条path
3772 20 : for (const auto& path : paths) {
3773 : // 检查是否是Clos网络（有nullptr接口）
3774 15 : bool isClos = false;
3775 18 : for (const auto& link : path.links) {
3776 : // fabric没有接口
3777 15 : if (link.GetSourceIface() == nullptr || link.GetTargetIface() == nullptr) {
3778 12 : isClos = true;
3779 12 : break;
3780 : }
3781 : }
3782 15 : if (!isClos) {
3783 : // Peer2Peer网络：直接处理每条link
3784 3 : HcclResult ret = InsertInnerLink(path, linkListVec);
3785 3 : if (ret != HCCL_SUCCESS)
3786 0 : return ret;
3787 : } else {
3788 : // Clos网络：找到peer2net和net2peer，组合成一条链路
3789 12 : HcclResult ret = InsertClosLinks(path, linkListVec);
3790 12 : if (ret != HCCL_SUCCESS)
3791 0 : return ret;
3792 : }
3793 : }
3794 5 : *linkList = linkListVec.data();
3795 5 : *listSize = linkListVec.size();
3796 5 : return HCCL_SUCCESS;
3797 5 : } catch (const InvalidParamsException& e) {
3798 0 : HCCL_ERROR(e.what());
3799 0 : return HCCL_E_PARA;
3800 0 : } catch (const NullPtrException& e) {
3801 0 : HCCL_ERROR(e.what());
3802 0 : return HCCL_E_PTR;
3803 0 : } catch (const std::exception& e) {
3804 0 : HCCL_ERROR(e.what());
3805 0 : return HCCL_E_INTERNAL;
3806 0 : }
3807 : }
3808 :
3809 3 : HcclResult CommunicatorImpl::GetTopoInstsByLayer(uint32_t netLayer, uint32_t **topoInsts, uint32_t *topoInstNum)
3810 : {
3811 : try {
3812 3 : CHK_PTR_NULL(rankGraph);
3813 3 : auto currNetType = rankGraph->GetNetType(netLayer);
3814 2 : if (currNetType != NetType::TOPO_FILE_DESC) {
3815 3 : HCCL_ERROR(
3816 : "[CommunicatorImpl::GetTopoInstsByLayer] Only support TOPO_FILE_DESC netType ,current netType is [%d]",
3817 : currNetType);
3818 1 : return HCCL_E_PARA;
3819 : }
3820 :
3821 1 : u32 num = 0;
3822 1 : rankGraph->GetTopoInstsByLayer(netLayer, topoInstsVec, num);
3823 :
3824 1 : *topoInsts = topoInstsVec.data();
3825 1 : *topoInstNum = topoInstsVec.size();
3826 :
3827 1 : return HCCL_SUCCESS;
3828 1 : } catch (const InvalidParamsException &e) {
3829 0 : HCCL_ERROR(e.what());
3830 0 : return HCCL_E_PARA;
3831 1 : } catch (const NullPtrException &e) {
3832 3 : HCCL_ERROR(e.what());
3833 1 : return HCCL_E_PTR;
3834 1 : } catch (const std::exception &e) {
3835 0 : HCCL_ERROR(e.what());
3836 0 : return HCCL_E_INTERNAL;
3837 0 : }
3838 : }
3839 :
3840 3 : HcclResult CommunicatorImpl::GetTopoType(uint32_t netLayer, uint32_t topoInstId, CommTopo* topoType)
3841 : {
3842 : try {
3843 3 : CHK_PTR_NULL(rankGraph);
3844 3 : auto currNetType = rankGraph->GetNetType(netLayer);
3845 2 : if (currNetType != NetType::TOPO_FILE_DESC) {
3846 3 : HCCL_ERROR(
3847 : "[CommunicatorImpl::GetTopoInstsByLayer] Only support TOPO_FILE_DESC netType, current netType is [%d]",
3848 : currNetType);
3849 1 : return HCCL_E_PARA;
3850 : }
3851 1 : Hccl::TopoType type;
3852 1 : HcclResult ret = rankGraph->GetTopoType(netLayer, topoInstId, type);
3853 1 : if (ret != HCCL_SUCCESS) {
3854 0 : HCCL_ERROR("[CommunicatorImpl::GetTopoType] Failed to get topo type at netLayer [%u] ret=%d", netLayer, ret);
3855 0 : return ret;
3856 : }
3857 : static const std::unordered_map<Hccl::TopoType, CommTopo> topoTypeMap = {
3858 : {Hccl::TopoType::CLOS, COMM_TOPO_CLOS},
3859 : {Hccl::TopoType::MESH_1D, COMM_TOPO_1DMESH},
3860 : {Hccl::TopoType::A3_SERVER, COMM_TOPO_910_93},
3861 3 : {Hccl::TopoType::A2_AX_SERVER, COMM_TOPO_A2AXSERVER}};
3862 1 : auto it = topoTypeMap.find(type);
3863 1 : if (it != topoTypeMap.end()) {
3864 1 : *topoType = it->second;
3865 1 : return HCCL_SUCCESS;
3866 : }
3867 0 : return HCCL_E_PARA;
3868 1 : } catch (const InvalidParamsException& e) {
3869 0 : HCCL_ERROR(e.what());
3870 0 : return HCCL_E_PARA;
3871 1 : } catch (const NullPtrException& e) {
3872 3 : HCCL_ERROR(e.what());
3873 1 : return HCCL_E_PTR;
3874 1 : } catch (const std::exception& e) {
3875 0 : HCCL_ERROR(e.what());
3876 0 : return HCCL_E_INTERNAL;
3877 0 : }
3878 : }
3879 :
3880 3 : HcclResult CommunicatorImpl::GetRanksByTopoInst(uint32_t netLayer, uint32_t topoInstId, uint32_t **ranks,
3881 : uint32_t *rankNum)
3882 : {
3883 : try {
3884 3 : CHK_PTR_NULL(rankGraph);
3885 3 : auto currNetType = rankGraph->GetNetType(netLayer);
3886 2 : if (currNetType != NetType::TOPO_FILE_DESC) {
3887 3 : HCCL_ERROR(
3888 : "[CommunicatorImpl::GetTopoInstsByLayer] Only support TOPO_FILE_DESC netType, current netType is [%d]",
3889 : currNetType);
3890 1 : return HCCL_E_PARA;
3891 : }
3892 1 : u32 num = 0;
3893 1 : auto ret = rankGraph->GetRanksByTopoInst(netLayer, topoInstId, ranksVec, num);
3894 1 : if (ret != HCCL_SUCCESS) {
3895 0 : HCCL_ERROR("[CommunicatorImpl::GetRanksByTopoInst] Failed to get topo type at netLayer [%u] ret=%d", netLayer, ret);
3896 0 : return ret;
3897 : }
3898 1 : *ranks = ranksVec.data();
3899 1 : *rankNum = ranksVec.size();
3900 1 : return HCCL_SUCCESS;
3901 1 : } catch (const InvalidParamsException &e) {
3902 0 : HCCL_ERROR(e.what());
3903 0 : return HCCL_E_PARA;
3904 1 : } catch (const NullPtrException &e) {
3905 3 : HCCL_ERROR(e.what());
3906 1 : return HCCL_E_PTR;
3907 1 : } catch (const std::exception &e) {
3908 0 : HCCL_ERROR(e.what());
3909 0 : return HCCL_E_INTERNAL;
3910 0 : }
3911 : }
3912 :
3913 1 : HcclResult CommunicatorImpl::GetInstSizeByNetLayer(uint32_t netLayer, uint32_t* rankNum)
3914 : {
3915 : try {
3916 1 : CHK_PTR_NULL(rankGraph);
3917 1 : u32 num = rankGraph->GetLocalInstSize(netLayer);
3918 1 : *rankNum = static_cast<uint32_t>(num);
3919 1 : return HCCL_SUCCESS;
3920 0 : } catch (const InvalidParamsException& e) {
3921 0 : HCCL_ERROR(e.what());
3922 0 : return HCCL_E_PARA;
3923 0 : } catch (const NullPtrException& e) {
3924 0 : HCCL_ERROR(e.what());
3925 0 : return HCCL_E_PTR;
3926 0 : } catch (const std::exception& e) {
3927 0 : HCCL_ERROR(e.what());
3928 0 : return HCCL_E_INTERNAL;
3929 0 : }
3930 : }
3931 :
3932 4 : HcclResult CommunicatorImpl::GetEndpointNum(uint32_t layer, uint32_t topoInstId, uint32_t* num)
3933 : {
3934 4 : CHK_PTR_NULL(rankGraph);
3935 4 : HcclResult ret = rankGraph->GetEndpointNum(layer, topoInstId, num);
3936 4 : if (ret != HCCL_SUCCESS) {
3937 0 : HCCL_ERROR("[CommunicatorImpl::GetEndpointNum] Faild to get endpoint num at netLayer [%u] with topoInstId[%u]", layer, topoInstId);
3938 0 : return ret;
3939 : }
3940 4 : return HCCL_SUCCESS;
3941 : }
3942 :
3943 3 : HcclResult CommunicatorImpl::GetEndpointDesc(uint32_t layer, uint32_t topoInstId, uint32_t *descNum, EndpointDesc *endpointDesc)
3944 : {
3945 3 : CHK_PTR_NULL(rankGraph);
3946 3 : HcclResult ret = rankGraph->GetEndpointDesc(layer, topoInstId, descNum, endpointDesc);
3947 3 : if (ret != HCCL_SUCCESS) {
3948 0 : HCCL_ERROR("[CommunicatorImpl::GetEndpointDesc] Failed to get endpoint desc at netLayer [%u] with descNum [%p]", layer, descNum);
3949 0 : return ret;
3950 : }
3951 3 : return HCCL_SUCCESS;
3952 : }
3953 :
3954 3 : HcclResult CommunicatorImpl::GetEndpointInfo(uint32_t rankId, const EndpointDesc* endPointDesc, EndpointAttr endpointAttr,
3955 : uint32_t infoLen, void* info)
3956 : {
3957 3 : CHK_PTR_NULL(rankGraph);
3958 3 : HcclResult ret = rankGraph->GetEndpointInfo(rankId, endPointDesc, endpointAttr, infoLen, info);
3959 3 : if (ret != HCCL_SUCCESS) {
3960 3 : HCCL_ERROR("[CommunicatorImpl::GetEndpointInfo] Faild to get endpoint info with rank [%u]", rankId);
3961 1 : return ret;
3962 : }
3963 2 : return HCCL_SUCCESS;
3964 : }
3965 :
3966 :
3967 5 : HcclResult CommunicatorImpl::SaveTopoDesc(std::string &identifier)
3968 : {
3969 5 : uint32_t topoType = 0;
3970 5 : CHK_RET(GetInstTopoTypeByNetLayer(0, &topoType)); // layer 0
3971 :
3972 5 : CommTopoDesc::GetInstance().SaveRankSize(identifier, rankSize);
3973 5 : CommTopoDesc::GetInstance().SaveL0TopoType(identifier, static_cast<CommTopo>(topoType));
3974 5 : return HCCL_SUCCESS;
3975 : }
3976 :
3977 3 : void CommunicatorImpl::CheckAcceleratorConsistency(AcceleratorState commAccelerator, AcceleratorState tilingAccelerator) const
3978 : {
3979 3 : bool isCommAiv = (commAccelerator == AcceleratorState::AIV || commAccelerator == AcceleratorState::AIV_ONLY);
3980 3 : bool isTilingCcu = (tilingAccelerator == AcceleratorState::CCU_MS || tilingAccelerator == AcceleratorState::CCU_SCHED);
3981 :
3982 3 : bool isCommCcu = (commAccelerator == AcceleratorState::CCU_MS || commAccelerator == AcceleratorState::CCU_SCHED);
3983 3 : bool isTilingAiv = (tilingAccelerator == AcceleratorState::AIV || tilingAccelerator == AcceleratorState::AIV_ONLY);
3984 :
3985 3 : if ((isCommAiv && isTilingCcu) || (isCommCcu && isTilingAiv)) {
3986 3 : HCCL_WARNING("CommunicatorImpl::GetTilingAccelerator comm accelerator is [%s] but tiling accelerator is [%s]",
3987 : commAccelerator.Describe().c_str(), tilingAccelerator.Describe().c_str());
3988 : }
3989 3 : }
3990 :
3991 7 : HcclResult CommunicatorImpl::GetTilingAccelerator(void *mc2Tiling, AcceleratorState& acceleratorState) const
3992 : {
3993 21 : HCCL_INFO("[CommunicatorImpl::%s] start.", __func__);
3994 7 : auto tilingVersion = *static_cast<uint32_t *>(mc2Tiling);
3995 21 : HCCL_INFO("[CommunicatorImpl:%s] Tiling version [%u]", __func__, tilingVersion);
3996 7 : if (tilingVersion != UNKNOWN_TILING_V1 && tilingVersion != UNKNOWN_TILING_V2) {
3997 3 : HCCL_ERROR("[CommunicatorImpl::GetTilingAccelerator] Tiling version not support, version[%u]", tilingVersion);
3998 1 : return HCCL_E_NOT_SUPPORT;
3999 : }
4000 6 : uint8_t accelerator{0};
4001 6 : if (tilingVersion == UNKNOWN_TILING_V1) {
4002 : // 从mc2Tiling中获取需要的算法信息,校验所有commConfig的communicationEngine是否一致
4003 6 : Mc2Tiling *mc2TilingPtr = reinterpret_cast<Mc2Tiling *>(mc2Tiling);
4004 6 : accelerator = static_cast<Mc2Tiling *>(mc2Tiling)->commConfig.communicationEngine;
4005 6 : Mc2CommConfig *commConfigPtr = reinterpret_cast<Mc2CommConfig *>(
4006 : reinterpret_cast<uint8_t *>(mc2TilingPtr) + sizeof(uint32_t) + sizeof(uint32_t) + sizeof(Mc2ServerCfg));
4007 12 : for (uint32_t index = 0; index < mc2TilingPtr->commConfigNum; index++) {
4008 6 : const Mc2CommConfig &commConfig = *(commConfigPtr + index);
4009 6 : if (commConfig.communicationEngine != accelerator) {
4010 0 : HCCL_ERROR("[CommunicatorImpl::GetTilingAccelerator] Input communicationEngine [%u] and [%u] not equal", commConfig.communicationEngine, accelerator);
4011 0 : return HCCL_E_PARA;
4012 : }
4013 : }
4014 : } else {
4015 0 : Mc2InitTilingInner *mc2TilingPtr = reinterpret_cast<Mc2InitTilingInner *>(mc2Tiling);
4016 0 : const auto offset = mc2TilingPtr->offset[0];
4017 0 : const auto &commConfig
4018 0 : = *(reinterpret_cast<const Mc2CcTilingInner *>(reinterpret_cast<const uint8_t *>(mc2TilingPtr) + offset));
4019 0 : accelerator = commConfig.communicationEngine;
4020 :
4021 0 : HCCL_INFO("[CommunicatorImpl::%s] tilingAccelerator[%u].", __func__, accelerator);
4022 : }
4023 :
4024 6 : HcclAccelerator hcclAccelerator = HcclAccelerator::DEFAULT;
4025 6 : if (accelerator <= HcclAccelerator::AICPU) {
4026 5 : hcclAccelerator = static_cast<HcclAccelerator::Value>(accelerator);
4027 : }
4028 18 : HCCL_INFO("[CommunicatorImpl::%s] hcclAccelerator[%s].", __func__, hcclAccelerator.Describe().c_str());
4029 6 : switch (hcclAccelerator) {
4030 1 : case HcclAccelerator::CCU_SCHED:
4031 1 : acceleratorState = AcceleratorState::CCU_SCHED;
4032 1 : break;
4033 1 : case HcclAccelerator::AIV:
4034 1 : acceleratorState = AcceleratorState::AIV;
4035 1 : break;
4036 0 : case HcclAccelerator::AIV_ONLY:
4037 0 : acceleratorState = AcceleratorState::AIV_ONLY;
4038 0 : break;
4039 4 : default:
4040 12 : HCCL_ERROR("[GetTilingAccelerator] Tiling hcclAccelerator not support, hcclAccelerator[%s]", hcclAccelerator.Describe().c_str());
4041 4 : return HCCL_E_NOT_SUPPORT;
4042 : }
4043 :
4044 2 : AcceleratorState commAccelerator = GetCommExecuteConfig().accState;
4045 2 : CheckAcceleratorConsistency(commAccelerator, acceleratorState);
4046 :
4047 2 : return HCCL_SUCCESS;
4048 : }
4049 :
4050 2 : HcclResult CommunicatorImpl::CalcNumBlocks(const CollOpParams &opParams, int32_t aivCoreLimit, std::string &algName,
4051 : u32 &numBlocks) const
4052 : {
4053 6 : HCCL_INFO("[CommunicatorImpl::CalcNumBlocks] count[%llu], dataType[%s], opType[%s], aivCoreLimit[%d], algName[%s].",
4054 : opParams.count, opParams.dataType.Describe().c_str(), opParams.opType.Describe().c_str(), aivCoreLimit,
4055 : algName.c_str());
4056 2 : numBlocks = aivCoreLimit;
4057 2 : return HCCL_SUCCESS;
4058 : }
4059 :
4060 2 : HcclResult CommunicatorImpl::GetAlgExecParam(const CollOpParams &opParams, bool clearEnable, void *&commContext,
4061 : u64 &len, u32 aivCoreLimit)
4062 : {
4063 6 : HCCL_INFO("[CommunicatorImpl::GetAlgExecParam] clearEnable[%d], aivCoreLimit[%u].", clearEnable, aivCoreLimit);
4064 2 : bool ifAiv = true;
4065 2 : std::string algName = "";
4066 2 : CHK_RET(HcomSelectAlg(opParams, aivCoreLimit, ifAiv, algName));
4067 2 : bool isAiv = (opExecuteConfig.accState == AcceleratorState::AIV || opExecuteConfig.accState == AcceleratorState::AIV_ONLY);
4068 2 : if (!isAiv) {
4069 3 : HCCL_WARNING("GetAlgExecParam only supported aiv.");
4070 1 : return HCCL_E_NOT_SUPPORT;
4071 : }
4072 :
4073 1 : u32 numBlocks = 0;
4074 1 : CHK_RET(CalcNumBlocks(opParams, aivCoreLimit, algName, numBlocks));
4075 :
4076 1 : return collService->GetAlgExecParam(clearEnable, numBlocks, commContext, len);
4077 2 : }
4078 :
4079 : HcclResult DeregisterOffloadSlaveStreams(const std::string &opTag);
4080 :
4081 1 : HcclResult CommunicatorImpl::ClearOpResource(const std::string &opTag)
4082 : {
4083 3 : HCCL_INFO("CommunicatorImpl::%s] opTag[%s]", __func__, opTag.c_str());
4084 : // 清空stream资源
4085 1 : CHK_RET(GetStreamManager().offload->ClearOpStream(opTag));
4086 : // 清空workspaceMem资源
4087 1 : offloadScrachBufferMap.erase(opTag);
4088 3 : HCCL_RUN_INFO("[CommunicatorImpl][%s] offloadScrachBuffer free, opTag[%s]", __func__, opTag.c_str());
4089 : // 清空input/output/scratch资源
4090 1 : CHK_RET(GetDataBufferManager().Deregister(opTag));
4091 1 : CHK_RET(GetLocalRmaBufManager().Dereg(opTag));
4092 : // 清空transport资源
4093 1 : CHK_RET(GetMemTransportManager()->ClearOpTransport(opTag));
4094 : // 清空aicpu_ts—host侧打包资源
4095 1 : CollServiceAiCpuImpl *aiCpuCollService = dynamic_cast<CollServiceAiCpuImpl *>(collServices[AcceleratorState::AICPU_TS].get());
4096 1 : CHK_PTR_NULL(aiCpuCollService);
4097 1 : CHK_RET(aiCpuCollService->ClearOpLoadedInfo(opTag));
4098 1 : return HCCL_SUCCESS;
4099 : }
4100 :
4101 47 : std::vector<LinkData> CommunicatorImpl::GetFullMeshLinks() const
4102 : {
4103 141 : HCCL_INFO("[CommunicatorImpl::%s] start.", __func__);
4104 :
4105 : // 遍历所有rank，两两建链
4106 47 : std::vector<LinkData> links;
4107 47 : std::unordered_set<LinkData> linkDataSet;
4108 47 : int rankSize = GetRankSize();
4109 47 : int myRank = GetMyRank();
4110 355 : for (int dRank = 0; dRank < rankSize; dRank++) {
4111 308 : if (myRank == dRank) {
4112 46 : continue;
4113 : }
4114 2102 : for (u32 level = 0; level < MAX_NET_LAYER; level++) {
4115 1872 : vector<LinkData> tempLinks;
4116 1872 : std::vector<NetInstance::Path> paths = GetRankGraph()->GetPaths(level, myRank, dRank);
4117 1904 : for (NetInstance::Path &path : paths) {
4118 32 : tempLinks.emplace_back(LinkData(path));
4119 : }
4120 :
4121 1872 : if (!tempLinks.empty()) {
4122 32 : linkDataSet.insert(tempLinks.at(0));
4123 32 : break;
4124 : }
4125 1904 : }
4126 : }
4127 :
4128 47 : links.assign(linkDataSet.begin(), linkDataSet.end());
4129 :
4130 141 : HCCL_INFO("[CommunicatorImpl::%s] end, links size[%zu]", __func__, links.size());
4131 47 : return links;
4132 47 : }
4133 :
4134 0 : ErrorMessageReport CommunicatorImpl::GetAicpuTaskException()
4135 : {
4136 0 : HcclResult ret = HCCL_SUCCESS;
4137 0 : ErrorMessageReport errorMessage;
4138 0 : if (kfcStatusTransferD2H != nullptr)
4139 : {
4140 0 : ret = kfcStatusTransferD2H->Get(sizeof(KfcStatus) + sizeof(KfcErrType),
4141 : sizeof(errorMessage), reinterpret_cast<uint8_t *>(&errorMessage));
4142 0 : if (ret != HCCL_SUCCESS)
4143 : {
4144 0 : HCCL_ERROR("GetAicpuTaskException get aicpu task exception failed.ret[%u]", ret);
4145 : }
4146 : } else {
4147 0 : HCCL_ERROR("GetAicpuTaskException kfcStatusTransferD2H is nullptr");
4148 : }
4149 0 : HCCL_INFO("[CommunicatorImpl::GetAicpuTaskException] end");
4150 0 : return errorMessage;
4151 : }
4152 :
4153 :
4154 0 : u32 CommunicatorImpl::GetRankInParentComm() const {
4155 0 : return static_cast<u32>(rankInParentComm);
4156 : }
4157 67 : void CommunicatorImpl::RegisterAicpuKernel()
4158 : {
4159 67 : aicpuKernelHolder_.Load();
4160 67 : }
4161 :
4162 6 : aclrtFuncHandle CommunicatorImpl::GetAicpuKernelFuncHandle(const char *kernelName) const
4163 : {
4164 6 : return aicpuKernelHolder_.GetAicpuKernelFuncHandle(kernelName);
4165 : }
4166 :
4167 0 : HcclResult CommunicatorImpl::Mc2AiCpuStreamAllocAndGetV2(rtStream_t *aiCpuStream)
4168 : {
4169 0 : if (aicpuStreamManager == nullptr)
4170 : {
4171 0 : HCCL_ERROR("[CommunicatorImpl::Mc2AiCpuStreamAllocAndGetV2] aicpuStreamManager is nullPtr!");
4172 0 : return HCCL_E_PTR;
4173 : }
4174 0 : aicpuStreamManager->AllocFreeStream();
4175 0 : Stream *stream = aicpuStreamManager->GetFreeStream();
4176 0 : *aiCpuStream = stream->GetPtr();
4177 0 : HCCL_RUN_INFO("[CommunicatorImpl::Mc2AiCpuStreamAllocAndGetV2] success, stream %s", stream->Describe().c_str());
4178 0 : return HCCL_SUCCESS;
4179 : }
4180 :
4181 2 : HcclResult CommunicatorImpl::SaveDpuStreamId()
4182 : {
4183 2 : dpuStreamId = HrtGetStreamId(dpuStream);
4184 6 : HCCL_INFO("[CommunicatorImpl::SaveDpuStreamId] dpuStreamId_[%u]", dpuStreamId);
4185 2 : return HCCL_SUCCESS;
4186 : }
4187 :
4188 0 : HcclResult CommunicatorImpl::GetRankIpPortMap(RankIpPortMapPtr& rankIpPortMap)
4189 : {
4190 0 : CHK_PTR_NULL(rankIpPortMap_);
4191 0 : rankIpPortMap = rankIpPortMap_;
4192 0 : return HCCL_SUCCESS;
4193 : }
4194 :
4195 0 : HcclResult CommunicatorImpl::SetRankIpPortMap(const RankIpPortMapPtr& rankIpPortMap)
4196 : {
4197 0 : CHK_PTR_NULL(rankIpPortMap);
4198 0 : rankIpPortMap_ = rankIpPortMap;
4199 0 : return HCCL_SUCCESS;
4200 : }
4201 :
4202 : } // namespace Hccl
|
