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