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